Claude Agent SDK

import asyncio

from claude_agent_sdk import (
    AssistantMessage,
    ClaudeAgentOptions,
    ResultMessage,
    TextBlock,
    ToolUseBlock,
    query,
)


async def main() -> None:
    options = ClaudeAgentOptions(
        cwd=".",  # agent 在哪个目录里工作
        allowed_tools=["Read", "Glob", "Grep"],  # 列在这里的工具会被自动批准，不弹审批
        max_turns=20,  # 防止失控
    )
    async for message in query(prompt="找出这个项目里所有的 TODO 注释，并按文件汇总", options=options):
        if isinstance(message, AssistantMessage):
            for block in message.content:  # 每个 AssistantMessage 只带一个内容块
                if isinstance(block, TextBlock):
                    print("Claude:", block.text)
                elif isinstance(block, ToolUseBlock):
                    print("调用工具:", block.name)
        elif isinstance(message, ResultMessage):  # 循环结束的标志
            print("结果类型:", message.subtype, "| 费用:", message.total_cost_usd)
            if message.subtype == "success":
                print(message.result)


asyncio.run(main())

import asyncio

from claude_agent_sdk import (
    ClaudeAgentOptions,
    HookContext,
    HookInput,
    HookJSONOutput,
    HookMatcher,
    PermissionResultAllow,
    PermissionResultDeny,
    ResultMessage,
    ToolPermissionContext,
    query,
)


async def can_use_tool(
    tool_name: str, input_data: dict, context: ToolPermissionContext
) -> PermissionResultAllow | PermissionResultDeny:
    print(f"\n工具：{tool_name}，参数：{input_data}")
    if input("允许吗？(y/n) ").lower() == "y":
        return PermissionResultAllow(updated_input=input_data)  # 也可以修改参数后再放行
    return PermissionResultDeny(message="用户拒绝了这个操作")  # Claude 能看到这条信息，会换个办法


async def keep_stream_open(
    input_data: HookInput, tool_use_id: str | None, context: HookContext
) -> HookJSONOutput:
    return {"continue_": True}  # 官方给出的变通写法：靠一个 hook 让流保持打开


async def prompt_stream():
    yield {"type": "user", "message": {"role": "user", "content": "在 /tmp 里创建一个测试文件，然后删掉它"}}


async def main() -> None:
    async for message in query(
        prompt=prompt_stream(),  # Python 中使用 can_use_tool 时需要“流式输入”
        options=ClaudeAgentOptions(
            can_use_tool=can_use_tool,
            hooks={"PreToolUse": [HookMatcher(matcher=None, hooks=[keep_stream_open])]},
        ),
    ):
        if isinstance(message, ResultMessage) and message.subtype == "success":
            print(message.result)


asyncio.run(main())

import asyncio
from typing import Annotated, Any

from claude_agent_sdk import ClaudeAgentOptions, ResultMessage, create_sdk_mcp_server, query, tool


@tool(
    "get_weather",
    "查询城市今天的天气",
    {"city": Annotated[str, "城市名，例如 北京"]},  # dict 写法：SDK 会转换成 JSON Schema
)
async def get_weather(args: dict[str, Any]) -> dict[str, Any]:
    data = {"北京": "晴，25°C", "上海": "小雨，22°C"}
    return {"content": [{"type": "text", "text": data.get(args["city"], "暂无数据")}]}


weather_server = create_sdk_mcp_server(name="weather", version="1.0.0", tools=[get_weather])


async def main() -> None:
    options = ClaudeAgentOptions(
        mcp_servers={"weather": weather_server},  # 这里的 key 就是 server_name
        allowed_tools=["mcp__weather__get_weather"],  # 预先批准，不弹审批
    )
    async for message in query(prompt="北京今天适合跑步吗？", options=options):
        if isinstance(message, ResultMessage) and message.subtype == "success":
            print(message.result)


asyncio.run(main())

import asyncio

from claude_agent_sdk import (
    AssistantMessage,
    ClaudeAgentOptions,
    ClaudeSDKClient,
    HookContext,
    HookInput,
    HookJSONOutput,
    HookMatcher,
    ResultMessage,
)


async def protect_env_files(
    input_data: HookInput, tool_use_id: str | None, context: HookContext
) -> HookJSONOutput:
    if input_data["hook_event_name"] != "PreToolUse":  # HookInput 是按事件名区分的联合类型
        return {}
    file_path = str(input_data["tool_input"].get("file_path", ""))
    if file_path.split("/")[-1] == ".env":
        return {
            "hookSpecificOutput": {
                "hookEventName": "PreToolUse",
                "permissionDecision": "deny",
                "permissionDecisionReason": "不允许修改 .env 文件",
            }
        }
    return {}  # 返回空对象表示放行


async def main() -> None:
    options = ClaudeAgentOptions(
        hooks={"PreToolUse": [HookMatcher(matcher="Write|Edit", hooks=[protect_env_files])]}
    )
    async with ClaudeSDKClient(options=options) as client:
        await client.query("创建一个 .env 文件，写入本地开发数据库配置")
        async for message in client.receive_response():
            if isinstance(message, (AssistantMessage, ResultMessage)):
                print(message)


asyncio.run(main())

import asyncio

from claude_agent_sdk import AgentDefinition, ClaudeAgentOptions, ResultMessage, query


async def main() -> None:
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Grep", "Glob", "Agent"],  # 主 agent 通过 Agent 工具调用 subagent
        agents={
            "code-reviewer": AgentDefinition(
                description="代码审查专家，负责质量、安全和可维护性审查。",  # Claude 根据它决定何时调用
                prompt="你是代码审查专家，关注安全漏洞、性能问题和编码规范，给出具体的改进建议。",
                tools=["Read", "Grep", "Glob"],  # 只读：只能分析，不能修改
                model="sonnet",
            ),
        },
    )
    async for message in query(prompt="审查 auth 模块的安全问题", options=options):
        if isinstance(message, ResultMessage) and message.subtype == "success":
            print(message.result)


asyncio.run(main())

import asyncio

from claude_agent_sdk import ClaudeAgentOptions, ClaudeSDKClient, ResultMessage


async def main() -> None:
    async with ClaudeSDKClient(options=ClaudeAgentOptions(allowed_tools=["Read", "Glob"])) as client:
        await client.query("分析一下 utils 目录是做什么的")
        async for msg in client.receive_response():
            if isinstance(msg, ResultMessage):
                print(msg.result)
        await client.query("基于刚才的分析，列出三个可以重构的点")  # 自动沿用同一个会话的上下文
        async for msg in client.receive_response():
            if isinstance(msg, ResultMessage):
                print(msg.result)


asyncio.run(main())