Python 版 Claude Code SDK 实战：从安装到自定义工具开发

权限控制与钩子

对于敏感操作，比如 Bash 命令执行，我们可以通过钩子进行细粒度拦截。例如，你可以阻止包含特定危险模式的命令。

async def check_bash_command(input_data, tool_use_id, context):
    tool_name = input_data["tool_name"]
    tool_input = input_data["tool_input"]
    
    if tool_name != "Bash":
        return {}
        
    command = tool_input.get("command", "")
    block_patterns = ["rm -rf", "foo.sh"]
    
    for pattern in block_patterns:
        if pattern in command:
            return {
                "hookSpecificOutput": {
                    "hookEventName": "PreToolUse",
                    "permissionDecision": "deny",
                    "permissionDecisionReason": f"Command contains invalid pattern: {pattern}",
                }
            }
    return {}

options = ClaudeAgentOptions(
    allowed_tools=["Bash"],
    hooks={
        "PreToolUse": [
            HookMatcher(matcher="Bash", hooks=[check_bash_command]),
        ],
    }
)

自定义工具开发

这是 SDK 最强大的部分之一。你不需要启动额外的进程，可以直接在 Python 进程中定义 MCP 服务器，这样性能更好且部署更简单。

定义工具函数

使用 @tool 装饰器标记你的异步函数。参数类型注解会自动映射为 JSON Schema。

from typing import Any
from claude_agent_sdk import tool, create_sdk_mcp_server, ClaudeAgentOptions, ClaudeSDKClient

# 定义一个简单的问候工具
@tool("greet", "Greet a user", {"name": str})
async def greet_user(args):
    return {
        "content": [
            {"type": "text", "text": f"Hello, {args['name']}!"}
        ]
    }

# 创建 SDK MCP 服务器
server = create_sdk_mcp_server(
    name="my-tools",
    version="1.0.0",
    tools=[greet_user]
)

# 在客户端中注册并使用
options = ClaudeAgentOptions(
    mcp_servers={"tools": server},
    allowed_tools=["mcp__tools__greet"]
)

async with ClaudeSDKClient(options=options) as client:
    await client.query("Greet Alice")
    async for msg in client.receive_response():
        print(msg)

计算器工具示例

下面是一个完整的计算器工具集实现，展示了如何组合多个工具到一个服务器中。

from typing import Any
from claude_agent_sdk import tool, create_sdk_mcp_server

# 加法
@tool("add", "Add two numbers", {"a": float, "b": float})
async def add_numbers(args: dict[str, Any]) -> dict[str, Any]:
    result = args["a"] + args["b"]
    return {"content": [{"type": "text", "text": f"{args['a']} + {args['b']} = {result}"}]}

# 减法
@tool("subtract", "Subtract one number from another", {"a": float, "b": float})
async def subtract_numbers(args: dict[str, Any]) -> dict[str, Any]:
    result = args["a"] - args["b"]
    return {"content": [{"type": "text", "text": f"{args['a']} - {args['b']} = {result}"}]}

# 更多工具如 multiply, divide 等...

# 组装服务器
calculator = create_sdk_mcp_server(
    name="calculator",
    tools=[add_numbers, subtract_numbers]  # 加上其他运算工具
)

options = ClaudeAgentOptions(
    mcp_servers={"calc": calculator},
    allowed_tools=[
        "mcp__calc__add",
        "mcp__calc__subtract",
        # ... 其他工具 ID
    ],
)

错误处理与异常捕获

SDK 定义了多种异常类型，方便我们针对不同场景做处理。比如找不到 CLI、连接失败或进程退出等情况。

from claude_agent_sdk import (
    ClaudeSDKError,
    CLINotFoundError,
    CLIConnectionError,
    ProcessError,
    CLIJSONDecodeError,
)

try:
    async for message in query(prompt="Hello"):
        pass
except CLINotFoundError:
    print("请安装 Claude Code")
except ProcessError as e:
    print(f"进程失败，退出码：{e.exit_code}")
except CLIJSONDecodeError as e:
    print(f"响应解析失败：{e}")

高级特性

流模式处理

通过 ClaudeSDKClient 的上下文管理器，可以实时接收并处理 Claude 的流式响应，这对于提升用户体验很有帮助。

async with ClaudeSDKClient(options=options) as client:
    await client.query("Tell me a story")
    async for message in client.receive_response():
        # 实时处理消息
        print(message)

混合 MCP 服务器

你甚至可以同时加载进程内 SDK 服务器和外部子进程服务器，实现更复杂的架构。

options = ClaudeAgentOptions(
    mcp_servers={
        "internal": sdk_server,  # 进程内 SDK 服务器
        "external": {            # 外部子进程服务器
            "type": "stdio",
            "command": "external-server"
        }
    }
)

迁移与总结

如果你是从旧版 SDK（<0.1.0）升级，需要注意一些命名变更，比如 ClaudeCodeOptions 已重命名为 ClaudeAgentOptions，并且系统提示配置方式有所调整。

总的来说，claude-code-sdk-python 为 Python 开发者提供了一套完善的工具链。无论是简单的问答集成，还是复杂的自定义工具编排，这套 SDK 都能胜任。掌握这些内容后，你就能构建出真正具备自主执行能力的 AI 应用了。