Claude Agent SDK Python 原始碼解讀：用 CLI 子程序架構橋接 Python 與 AI 代理

src/claude_agent_sdk/
├── __init__.py                    # 公開 API 彙出 + MCP Server 工廠
├── query.py                       # query() 單次查詢公開介面
├── client.py                      # ClaudeSDKClient 互動式客戶端
├── types.py                       # 完整型別定義（500+ 行）
├── _errors.py                     # 階層式錯誤類別
├── _version.py                    # SDK 版本
├── _cli_version.py                # 綁定的 CLI 版本
├── _bundled/                      # 綁定的 CLI 執行檔
└── _internal/                     # 內部實作層
    ├── __init__.py
    ├── client.py                  # InternalClient 協調器
    ├── query.py                   # Query 控制協定處理
    ├── message_parser.py          # 訊息解析器
    └── transport/
        ├── __init__.py            # Transport 抽象基類
        └── subprocess_cli.py      # CLI 子程序 Transport 實作
Code language: PHP (php)

class Transport(ABC):
    @abstractmethod
    async def connect(self) -> None: ...

    @abstractmethod
    async def write(self, data: str) -> None: ...

    @abstractmethod
    def read_messages(self) -> AsyncIterator[dict[str, Any]]: ...

    @abstractmethod
    async def close(self) -> None: ...

    @abstractmethod
    def is_ready(self) -> bool: ...

    @abstractmethod
    async def end_input(self) -> None: ...
Code language: CSS (css)

class SubprocessCLITransport(Transport):
    def __init__(self, prompt, options):
        self._prompt = prompt
        self._is_streaming = not isinstance(prompt, str)
        # ...
        self._max_buffer_size = (
            options.max_buffer_size
            if options.max_buffer_size is not None
            else _DEFAULT_MAX_BUFFER_SIZE  # 1MB
        )
        self._write_lock: anyio.Lock = anyio.Lock()

async def _read_messages_impl(self) -> AsyncIterator[dict[str, Any]]:
    json_buffer = ""
    try:
        async for line in self._stdout_stream:
            line_str = line.strip()
            if not line_str:
                continue
            json_lines = line_str.split("\n")
            for json_line in json_lines:
                json_buffer += json_line
                if len(json_buffer) > self._max_buffer_size:
                    json_buffer = ""
                    raise SDKJSONDecodeError(...)
                try:
                    data = json.loads(json_buffer)
                    json_buffer = ""
                    yield data
                except json.JSONDecodeError:
                    continue  # 繼續累積，等待完整 JSON
Code language: PHP (php)

async def query(
    *,
    prompt: str | AsyncIterable[dict[str, Any]],
    options: ClaudeAgentOptions | None = None,
    transport: Transport | None = None,
) -> AsyncIterator[Message]:
    if options is None:
        options = ClaudeAgentOptions()
    os.environ["CLAUDE_CODE_ENTRYPOINT"] = "sdk-py"
    client = InternalClient()
    async for message in client.process_query(
        prompt=prompt, options=options, transport=transport
    ):
        yield message
Code language: JavaScript (javascript)

class ClaudeSDKClient:
    async def connect(self, prompt=None) -> None:
        # Auto-connect with empty async iterable
        async def _empty_stream() -> AsyncIterator[dict[str, Any]]:
            return
            yield {}  # type: ignore[unreachable]
        actual_prompt = _empty_stream() if prompt is None else prompt
        # ...

class Query:
    def __init__(self, transport, is_streaming_mode, can_use_tool, hooks, sdk_mcp_servers, ...):
        self.pending_control_responses: dict[str, anyio.Event] = {}
        self.pending_control_results: dict[str, dict[str, Any] | Exception] = {}
        self.hook_callbacks: dict[str, Callable[..., Any]] = {}
        self._message_send, self._message_receive = anyio.create_memory_object_stream[
            dict[str, Any]
        ](max_buffer_size=100)

async def _read_messages(self) -> None:
    async for message in self.transport.read_messages():
        msg_type = message.get("type")
        if msg_type == "control_response":
            # 路由到等待中的控制請求
            response = message.get("response", {})
            request_id = response.get("request_id")
            if request_id in self.pending_control_responses:
                event = self.pending_control_responses[request_id]
                self.pending_control_results[request_id] = response
                event.set()
            continue
        elif msg_type == "control_request":
            # CLI 發來的控制請求，在 task group 中處理
            self._tg.start_soon(self._handle_control_request, request)
            continue
        # 普通 SDK 訊息放入佇列
        await self._message_send.send(message)
Code language: PHP (php)

class SyncHookJSONOutput(TypedDict):
    # Using continue_ to avoid Python keyword (converted to "continue" for CLI)
    continue_: NotRequired[bool]
    suppressOutput: NotRequired[bool]
    stopReason: NotRequired[str]
    # ...
    hookSpecificOutput: NotRequired[HookSpecificOutput]
Code language: CSS (css)

def _convert_hook_output_for_cli(hook_output: dict[str, Any]) -> dict[str, Any]:
    converted = {}
    for key, value in hook_output.items():
        if key == "async_":
            converted["async"] = value
        elif key == "continue_":
            converted["continue"] = value
        else:
            converted[key] = value
    return converted
Code language: PHP (php)

@dataclass
class SdkMcpTool(Generic[T]):
    name: str
    description: str
    input_schema: type[T] | dict[str, Any]
    handler: Callable[[T], Awaitable[dict[str, Any]]]

def tool(name, description, input_schema):
    def decorator(handler):
        return SdkMcpTool(
            name=name, description=description,
            input_schema=input_schema, handler=handler,
        )
    return decorator
Code language: CSS (css)

def create_sdk_mcp_server(name, version="1.0.0", tools=None):
    server = Server(name, version=version)
    if tools:
        tool_map = {tool_def.name: tool_def for tool_def in tools}

        @server.list_tools()
        async def list_tools() -> list[Tool]:
            # 自動將 Python 型別轉換為 JSON Schema
            for param_name, param_type in tool_def.input_schema.items():
                if param_type is str:
                    properties[param_name] = {"type": "string"}
                elif param_type is int:
                    properties[param_name] = {"type": "integer"}
                # ...

        @server.call_tool()
        async def call_tool(name, arguments):
            result = await tool_def.handler(arguments)
            # 將結果轉為 MCP TextContent/ImageContent

    return McpSdkServerConfig(type="sdk", name=name, instance=server)
Code language: PHP (php)

async def _handle_sdk_mcp_request(self, server_name, message):
    server = self.sdk_mcp_servers[server_name]
    method = message.get("method")

    # TODO: Python MCP SDK lacks the Transport abstraction that TypeScript has.
    # TypeScript: server.connect(transport) allows custom transports
    # Python: server.run(read_stream, write_stream) requires actual streams
    #
    # This forces us to manually route methods.

    if method == "tools/list":
        handler = server.request_handlers.get(ListToolsRequest)
        result = await handler(request)
        # 手動轉換回 JSONRPC 格式
Code language: PHP (php)

@dataclass
class PermissionResultAllow:
    behavior: Literal["allow"] = "allow"
    updated_input: dict[str, Any] | None = None
    updated_permissions: list[PermissionUpdate] | None = None

@dataclass
class PermissionResultDeny:
    behavior: Literal["deny"] = "deny"
    message: str = ""
    interrupt: bool = False

PermissionResult = PermissionResultAllow | PermissionResultDeny
Code language: CSS (css)

def _find_cli(self) -> str:
    # 優先使用 bundled CLI
    bundled_cli = self._find_bundled_cli()
    if bundled_cli:
        return bundled_cli
    # 其次用 PATH 中的 claude
    if cli := shutil.which("claude"):
        return cli
    # 最後檢查常見安裝位置
    locations = [
        Path.home() / ".npm-global/bin/claude",
        Path("/usr/local/bin/claude"),
        Path.home() / ".local/bin/claude",
        Path.home() / "node_modules/.bin/claude",
        Path.home() / ".yarn/bin/claude",
        Path.home() / ".claude/local/claude",
    ]
Code language: PHP (php)

# Hook 輸入使用 discriminated union
HookInput = (
    PreToolUseHookInput
    | PostToolUseHookInput
    | PostToolUseFailureHookInput
    | UserPromptSubmitHookInput
    | StopHookInput
    | SubagentStopHookInput
    | PreCompactHookInput
)

# 每個變體都有 Literal 鑑別欄位
class PreToolUseHookInput(BaseHookInput):
    hook_event_name: Literal["PreToolUse"]
    tool_name: str
    tool_input: dict[str, Any]

async def stream_input(self, stream: AsyncIterable[dict[str, Any]]) -> None:
    try:
        async for message in stream:
            if self._closed:
                break
            await self.transport.write(json.dumps(message) + "\n")

        # 有 MCP Server 或 Hooks 時，等待第一個 result 才關 stdin
        has_hooks = bool(self.hooks)
        if self.sdk_mcp_servers or has_hooks:
            try:
                with anyio.move_on_after(self._stream_close_timeout):
                    await self._first_result_event.wait()
            except Exception:
                pass  # 超時也繼續
        await self.transport.end_input()
Code language: PHP (php)