From 0def306d1e1c84e8590a481a9eaf786c6fbc6fd4 Mon Sep 17 00:00:00 2001 From: zion Date: Fri, 1 May 2026 21:18:38 +0600 Subject: [PATCH 1/2] feat(types): add SystemPromptBlocks and accept list-form system_prompt in ClaudeAgentOptions --- src/claude_agent_sdk/types.py | 19 ++++++++++++++++--- tests/test_types.py | 13 +++++++++++++ 2 files changed, 29 insertions(+), 3 deletions(-) diff --git a/src/claude_agent_sdk/types.py b/src/claude_agent_sdk/types.py index 9c2be63fe..284c098dd 100644 --- a/src/claude_agent_sdk/types.py +++ b/src/claude_agent_sdk/types.py @@ -59,6 +59,15 @@ class SystemPromptFile(TypedDict): path: str +SystemPromptBlocks = list[dict[str, Any]] +"""Anthropic-compatible list form for the Messages API ``system`` parameter. + +The SDK forwards these blocks as structured JSON instead of flattening them +into a string so cache-control metadata and other per-block fields are +preserved. +""" + + class TaskBudget(TypedDict): """API-side task budget in tokens. @@ -1478,14 +1487,18 @@ class ClaudeAgentOptions: To restrict which tools are available at all, use ``tools``. """ - system_prompt: str | SystemPromptPreset | SystemPromptFile | None = None + system_prompt: ( + str | SystemPromptBlocks | SystemPromptPreset | SystemPromptFile | None + ) = None """System prompt configuration. - ``str`` — Use a custom system prompt. + - ``list[dict[str, Any]]`` — Forward structured Anthropic ``system`` content + blocks as-is. - ``{"type": "preset", "preset": "claude_code"}`` — Use Claude Code's default - system prompt. + system prompt. - ``{"type": "preset", "preset": "claude_code", "append": "..."}`` — Default - prompt with appended instructions. + prompt with appended instructions. """ mcp_servers: dict[str, McpServerConfig] | str | Path = field(default_factory=dict) diff --git a/tests/test_types.py b/tests/test_types.py index fbd07509f..b991ef515 100644 --- a/tests/test_types.py +++ b/tests/test_types.py @@ -177,6 +177,19 @@ def test_claude_code_options_with_system_prompt_file(self): "path": "/path/to/prompt.md", } + def test_claude_code_options_with_system_prompt_blocks(self): + """Test Options with structured system prompt blocks.""" + blocks = [ + {"type": "text", "text": "Be helpful."}, + { + "type": "text", + "text": "Use cached reference material.", + "cache_control": {"type": "ephemeral"}, + }, + ] + options = ClaudeAgentOptions(system_prompt=blocks) + assert options.system_prompt == blocks + def test_claude_code_options_with_session_continuation(self): """Test Options with session continuation.""" options = ClaudeAgentOptions(continue_conversation=True, resume="session-123") From 84bd50934cff2b3c80e874817ed8baaabc009be1 Mon Sep 17 00:00:00 2001 From: zion Date: Fri, 1 May 2026 21:18:58 +0600 Subject: [PATCH 2/2] fix(transport): handle list-form system_prompt by materializing JSON to temp file --- .../_internal/transport/subprocess_cli.py | 51 +++++++++++++++++-- tests/test_transport.py | 29 +++++++++++ 2 files changed, 75 insertions(+), 5 deletions(-) diff --git a/src/claude_agent_sdk/_internal/transport/subprocess_cli.py b/src/claude_agent_sdk/_internal/transport/subprocess_cli.py index 9a1d74580..1eb618615 100644 --- a/src/claude_agent_sdk/_internal/transport/subprocess_cli.py +++ b/src/claude_agent_sdk/_internal/transport/subprocess_cli.py @@ -6,6 +6,7 @@ import platform import re import shutil +import tempfile from collections.abc import AsyncIterable, AsyncIterator from contextlib import suppress from pathlib import Path @@ -58,6 +59,7 @@ def __init__( if options.max_buffer_size is not None else _DEFAULT_MAX_BUFFER_SIZE ) + self._generated_system_prompt_file: Path | None = None self._write_lock: anyio.Lock = anyio.Lock() def _find_cli(self) -> str: @@ -200,24 +202,59 @@ def _apply_skills_defaults( return allowed_tools, setting_sources + def _materialize_system_prompt_blocks(self, blocks: list[dict[str, Any]]) -> str: + """Persist structured system prompt blocks for CLI handoff.""" + if self._generated_system_prompt_file is None: + with tempfile.NamedTemporaryFile( + mode="w", + encoding="utf-8", + prefix="claude-agent-sdk-system-prompt-", + suffix=".json", + delete=False, + ) as temp_file: + json.dump(blocks, temp_file) + self._generated_system_prompt_file = Path(temp_file.name) + + return str(self._generated_system_prompt_file) + + def _cleanup_generated_system_prompt_file(self) -> None: + """Remove any temporary system prompt file created by the SDK.""" + path = self._generated_system_prompt_file + self._generated_system_prompt_file = None + if path is not None: + with suppress(FileNotFoundError): + path.unlink() + def _build_command(self) -> list[str]: """Build CLI command with arguments.""" if self._cli_path is None: raise CLINotFoundError("CLI path not resolved. Call connect() first.") cmd = [self._cli_path, "--output-format", "stream-json", "--verbose"] - if self._options.system_prompt is None: + sp = self._options.system_prompt + if sp is None: cmd.extend(["--system-prompt", ""]) - elif isinstance(self._options.system_prompt, str): - cmd.extend(["--system-prompt", self._options.system_prompt]) - else: - sp = self._options.system_prompt + elif isinstance(sp, str): + cmd.extend(["--system-prompt", sp]) + elif isinstance(sp, list): + cmd.extend( + [ + "--system-prompt-file", + self._materialize_system_prompt_blocks(sp), + ] + ) + elif isinstance(sp, dict): if sp.get("type") == "file": cmd.extend(["--system-prompt-file", cast(SystemPromptFile, sp)["path"]]) elif sp.get("type") == "preset" and "append" in sp: cmd.extend( ["--append-system-prompt", cast(SystemPromptPreset, sp)["append"]] ) + else: + raise TypeError( + "system_prompt must be None, a string, a system prompt preset/file " + "mapping, or a list of Anthropic content blocks" + ) # Handle tools option (base set of tools) if self._options.tools is not None: @@ -475,6 +512,7 @@ async def connect(self) -> None: self._ready = True except FileNotFoundError as e: + self._cleanup_generated_system_prompt_file() # Check if the error comes from the working directory or the CLI if self._cwd and not Path(self._cwd).exists(): error = CLIConnectionError( @@ -486,6 +524,7 @@ async def connect(self) -> None: self._exit_error = error raise error from e except Exception as e: + self._cleanup_generated_system_prompt_file() error = CLIConnectionError(f"Failed to start Claude Code: {e}") self._exit_error = error raise error from e @@ -513,6 +552,7 @@ async def close(self) -> None: """Close the transport and clean up resources.""" if not self._process: self._ready = False + self._cleanup_generated_system_prompt_file() return # Cancel stderr reader if active @@ -562,6 +602,7 @@ async def close(self) -> None: self._stdin_stream = None self._stderr_stream = None self._exit_error = None + self._cleanup_generated_system_prompt_file() async def write(self, data: str) -> None: """Write raw data to the transport.""" diff --git a/tests/test_transport.py b/tests/test_transport.py index efe0e2c9d..f9f5159a6 100644 --- a/tests/test_transport.py +++ b/tests/test_transport.py @@ -1,8 +1,10 @@ """Tests for Claude SDK transport layer.""" +import json import os import uuid from contextlib import nullcontext +from pathlib import Path from unittest.mock import AsyncMock, MagicMock, patch import anyio @@ -150,6 +152,33 @@ def test_build_command_with_system_prompt_file(self): assert "--system-prompt-file" in cmd assert "/path/to/prompt.md" in cmd + def test_build_command_with_system_prompt_blocks(self): + """Test building CLI command with structured system prompt blocks.""" + blocks = [ + {"type": "text", "text": "Be helpful."}, + { + "type": "text", + "text": "Use cached reference material.", + "cache_control": {"type": "ephemeral"}, + }, + ] + transport = SubprocessCLITransport( + prompt="test", + options=make_options(system_prompt=blocks), + ) + + cmd = transport._build_command() + assert "--system-prompt" not in cmd + assert "--append-system-prompt" not in cmd + assert "--system-prompt-file" in cmd + + system_prompt_path = Path(cmd[cmd.index("--system-prompt-file") + 1]) + assert system_prompt_path.exists() + assert json.loads(system_prompt_path.read_text(encoding="utf-8")) == blocks + + anyio.run(transport.close) + assert not system_prompt_path.exists() + def test_build_command_with_options(self): """Test building CLI command with options.""" transport = SubprocessCLITransport(