阶段 7 · Agent SDK——用 Claude Code 引擎构建你自己的代理

1
你的代码 → query() → 子进程 claude CLI
2
                         ↓
3
              Claude 接收 prompt + 工具定义
4
                         ↓
5
                  评估 → 决策
6
                    ↙         ↘
7
              文本响应        工具调用
8
                              ↓
9
                         SDK 执行工具
10
                              ↓
11
                       结果反馈给 Claude
12
                              ↓
13
                    重复，直到无工具调用
14
                         ↓
15
                    返回 ResultMessage

1
# Summary instructions
2
When summarizing this conversation, always preserve:
3
- The current task objective and acceptance criteria
4
- File paths that have been read or modified
5
- Test results and error messages

1
disallowedTools（裸名）→ 从 Claude 上下文中移除该工具
2
disallowedTools（作用域规则）→ 工具可见，但阻止匹配的调用
3
allowedTools → 预批准，调用时不弹权限确认
4
permissionMode → 处理所有不在上面两条规则内的工具

1
# Python：纯字符串
2
options = ClaudeAgentOptions(
3
    system_prompt="你是一名资深 Python 工程师，严格遵守 PEP 8。"
4
)
5

6
# Python：在 Claude Code 预设基础上追加
7
options = ClaudeAgentOptions(
8
    system_prompt={"type": "preset", "preset": "claude_code", "append": "额外约束..."}
9
)

1
// TypeScript
2
options: {
3
  systemPrompt: "你是一名资深 TypeScript 工程师，所有函数必须有类型注解。"
4
}
5
// 或追加到 Claude Code 预设
6
options: {
7
  systemPrompt: { type: "preset", preset: "claude_code", append: "额外约束..." }
8
}

1
import asyncio
2
from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, ResultMessage
3

4

5
async def main():
6
    async for message in query(
7
        prompt="列出当前目录下所有 .py 文件，并统计总行数。",
8
        options=ClaudeAgentOptions(
9
            allowed_tools=["Bash", "Glob"],
10
            permission_mode="acceptEdits",
11
            max_turns=10,
12
            max_budget_usd=0.50,
13
        ),
14
    ):
15
        if isinstance(message, AssistantMessage):
16
            for block in message.content:
17
                if hasattr(block, "text"):
18
                    print(block.text, end="")
19
                elif hasattr(block, "name"):
20
                    print(f"\n[调用工具: {block.name}]")
21
        elif isinstance(message, ResultMessage):
22
            if message.subtype == "success":
23
                print(f"\n\n完成。花费：${message.total_cost_usd:.4f}")
24
            else:
25
                print(f"\n停止原因：{message.subtype}")
26

27

28
asyncio.run(main())

1
import { query } from "@anthropic-ai/claude-agent-sdk";
2

3
for await (const message of query({
4
  prompt: "列出当前目录下所有 .ts 文件，并统计总行数。",
5
  options: {
6
    allowedTools: ["Bash", "Glob"],
7
    permissionMode: "acceptEdits",
8
    maxTurns: 10,
9
    maxBudgetUsd: 0.50,
10
  },
11
})) {
12
  if (message.type === "assistant") {
13
    for (const block of message.message.content) {
14
      if ("text" in block) process.stdout.write(block.text);
15
      else if ("name" in block) console.log(`\n[调用工具: ${block.name}]`);
16
    }
17
  } else if (message.type === "result") {
18
    if (message.subtype === "success") {
19
      console.log(`\n\n完成。花费：$${message.total_cost_usd.toFixed(4)}`);
20
    } else {
21
      console.log(`\n停止原因：${message.subtype}`);
22
    }
23
  }
24
}

1
import asyncio
2
import json
3
from typing import Any
4
import httpx
5
from claude_agent_sdk import (
6
    query,
7
    ClaudeAgentOptions,
8
    ResultMessage,
9
    tool,
10
    create_sdk_mcp_server,
11
    ToolAnnotations,
12
)
13

14

15
# 1. 用 @tool 装饰器定义工具
16
@tool(
17
    "get_weather",
18
    "获取指定经纬度的当前气温（摄氏度）",
19
    {"latitude": float, "longitude": float},
20
    annotations=ToolAnnotations(readOnlyHint=True),  # 只读，可并行调用
21
)
22
async def get_weather(args: dict[str, Any]) -> dict[str, Any]:
23
    async with httpx.AsyncClient() as client:
24
        resp = await client.get(
25
            "https://api.open-meteo.com/v1/forecast",
26
            params={
27
                "latitude": args["latitude"],
28
                "longitude": args["longitude"],
29
                "current": "temperature_2m",
30
                "temperature_unit": "celsius",
31
            },
32
        )
33
    data = resp.json()
34
    temp = data["current"]["temperature_2m"]
35
    return {"content": [{"type": "text", "text": f"当前气温：{temp}°C"}]}
36

37

38
@tool(
39
    "calculate",
40
    "执行基础四则运算",
41
    {
42
        "type": "object",
43
        "properties": {
44
            "expression": {"type": "string", "description": "数学表达式，如 '2 + 3 * 4'"},
45
        },
46
        "required": ["expression"],
47
    },
48
)
49
async def calculate(args: dict[str, Any]) -> dict[str, Any]:
50
    try:
51
        # 仅允许安全的数学字符
52
        expr = args["expression"]
53
        allowed = set("0123456789+-*/()., ")
54
        if not all(c in allowed for c in expr):
55
            return {
56
                "content": [{"type": "text", "text": "表达式包含不允许的字符"}],
57
                "is_error": True,
58
            }
59
        result = eval(expr)  # noqa: S307 — 已做字符白名单过滤
60
        return {"content": [{"type": "text", "text": f"结果：{result}"}]}
61
    except Exception as e:
62
        return {
63
            "content": [{"type": "text", "text": f"计算失败：{e}"}],
64
            "is_error": True,
65
        }
66

67

68
# 2. 把工具注册到进程内 MCP 服务器
69
tools_server = create_sdk_mcp_server(
70
    name="mytools",
71
    version="1.0.0",
72
    tools=[get_weather, calculate],
73
)
74

75

76
async def main():
77
    async for message in query(
78
        prompt="北京（纬度 39.9，经度 116.4）现在多少度？顺便算一下 (39.9 + 116.4) * 2",
79
        options=ClaudeAgentOptions(
80
            mcp_servers={"mytools": tools_server},
81
            # 通配符允许所有 mytools 下的工具
82
            allowed_tools=["mcp__mytools__*"],
83
            # 不允许 Claude 使用内置 Bash 工具
84
            tools=[],  # 清空内置工具，仅使用自定义工具
85
        ),
86
    ):
87
        if isinstance(message, ResultMessage) and message.subtype == "success":
88
            print(message.result)
89

90

91
asyncio.run(main())

1
import { query, tool, createSdkMcpServer } from "@anthropic-ai/claude-agent-sdk";
2
import { z } from "zod";
3

4
// 1. 定义工具（Zod schema 自动推导 handler 参数类型）
5
const getWeather = tool(
6
  "get_weather",
7
  "获取指定经纬度的当前气温（摄氏度）",
8
  {
9
    latitude: z.number().describe("纬度"),
10
    longitude: z.number().describe("经度"),
11
  },
12
  async (args) => {
13
    const resp = await fetch(
14
      `https://api.open-meteo.com/v1/forecast?latitude=${args.latitude}&longitude=${args.longitude}&current=temperature_2m&temperature_unit=celsius`
15
    );
16
    const data: any = await resp.json();
17
    return {
18
      content: [{ type: "text", text: `当前气温：${data.current.temperature_2m}°C` }],
19
    };
20
  },
21
  { annotations: { readOnlyHint: true } }
22
);
23

24
const calculate = tool(
25
  "calculate",
26
  "执行基础四则运算",
27
  {
28
    expression: z.string().describe("数学表达式，如 '2 + 3 * 4'"),
29
  },
30
  async (args) => {
31
    const allowed = /^[0-9+\-*/().,\s]+$/;
32
    if (!allowed.test(args.expression)) {
33
      return {
34
        content: [{ type: "text", text: "表达式包含不允许的字符" }],
35
        isError: true,
36
      };
37
    }
38
    try {
39
      const result = Function(`"use strict"; return (${args.expression})`)();
40
      return { content: [{ type: "text", text: `结果：${result}` }] };
41
    } catch (e) {
42
      return { content: [{ type: "text", text: `计算失败：${e}` }], isError: true };
43
    }
44
  }
45
);
46

47
// 2. 创建进程内 MCP 服务器
48
const toolsServer = createSdkMcpServer({
49
  name: "mytools",
50
  version: "1.0.0",
51
  tools: [getWeather, calculate],
52
});
53

54
// 3. 注入到 query
55
for await (const message of query({
56
  prompt: "北京（纬度 39.9，经度 116.4）现在多少度？顺便算一下 (39.9 + 116.4) * 2",
57
  options: {
58
    mcpServers: { mytools: toolsServer },
59
    allowedTools: ["mcp__mytools__*"],
60
    tools: [],  // 清空内置工具
61
  },
62
})) {
63
  if (message.type === "result" && message.subtype === "success") {
64
    console.log(message.result);
65
  }
66
}

1
import asyncio
2
from pydantic import BaseModel
3
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage
4

5

6
class BugReport(BaseModel):
7
    file_path: str
8
    line_number: int
9
    severity: str          # "low" | "medium" | "high"
10
    description: str
11
    suggested_fix: str
12

13

14
class CodeReview(BaseModel):
15
    bugs: list[BugReport]
16
    overall_quality: str   # "good" | "needs_improvement" | "critical"
17
    summary: str
18

19

20
async def main():
21
    async for message in query(
22
        prompt="Review the auth.py file for bugs and security issues.",
23
        options=ClaudeAgentOptions(
24
            allowed_tools=["Read", "Glob"],
25
            output_format={
26
                "type": "json_schema",
27
                "schema": CodeReview.model_json_schema(),
28
            },
29
        ),
30
    ):
31
        if isinstance(message, ResultMessage):
32
            if message.subtype == "success" and message.structured_output:
33
                review = CodeReview.model_validate(message.structured_output)
34
                print(f"整体质量：{review.overall_quality}")
35
                print(f"摘要：{review.summary}")
36
                for bug in review.bugs:
37
                    print(f"  [{bug.severity}] {bug.file_path}:{bug.line_number} — {bug.description}")
38
            elif message.subtype == "error_max_structured_output_retries":
39
                print("结构化输出生成失败，尝试简化 schema 或提示词。")
40

41

42
asyncio.run(main())

1
import { query } from "@anthropic-ai/claude-agent-sdk";
2
import { z } from "zod";
3

4
const BugReport = z.object({
5
  file_path: z.string(),
6
  line_number: z.number(),
7
  severity: z.enum(["low", "medium", "high"]),
8
  description: z.string(),
9
  suggested_fix: z.string(),
10
});
11

12
const CodeReview = z.object({
13
  bugs: z.array(BugReport),
14
  overall_quality: z.enum(["good", "needs_improvement", "critical"]),
15
  summary: z.string(),
16
});
17

18
type CodeReview = z.infer<typeof CodeReview>;
19

20
for await (const message of query({
21
  prompt: "Review the auth.ts file for bugs and security issues.",
22
  options: {
23
    allowedTools: ["Read", "Glob"],
24
    outputFormat: {
25
      type: "json_schema",
26
      schema: z.toJSONSchema(CodeReview),
27
    },
28
  },
29
})) {
30
  if (message.type === "result") {
31
    if (message.subtype === "success" && message.structured_output) {
32
      const parsed = CodeReview.safeParse(message.structured_output);
33
      if (parsed.success) {
34
        const review: CodeReview = parsed.data;
35
        console.log(`整体质量：${review.overall_quality}`);
36
        review.bugs.forEach((bug) =>
37
          console.log(`  [${bug.severity}] ${bug.file_path}:${bug.line_number} — ${bug.description}`)
38
        );
39
      }
40
    } else if (message.subtype === "error_max_structured_output_retries") {
41
      console.error("结构化输出生成失败");
42
    }
43
  }
44
}

1
import asyncio
2
import sys
3
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage
4
from claude_agent_sdk.types import StreamEvent
5

6

7
async def stream_with_progress():
8
    options = ClaudeAgentOptions(
9
        include_partial_messages=True,
10
        allowed_tools=["Read", "Glob", "Grep"],
11
        max_turns=15,
12
    )
13

14
    in_tool = False  # 跟踪当前是否在工具调用内
15

16
    async for message in query(
17
        prompt="在代码库里找出所有 TODO 注释，按文件汇总。", options=options
18
    ):
19
        if isinstance(message, StreamEvent):
20
            event = message.event
21
            etype = event.get("type")
22

23
            if etype == "content_block_start":
24
                block = event.get("content_block", {})
25
                if block.get("type") == "tool_use":
26
                    print(f"\n⟳ {block.get('name')}...", end="", flush=True)
27
                    in_tool = True
28

29
            elif etype == "content_block_delta":
30
                delta = event.get("delta", {})
31
                if delta.get("type") == "text_delta" and not in_tool:
32
                    sys.stdout.write(delta.get("text", ""))
33
                    sys.stdout.flush()
34

35
            elif etype == "content_block_stop" and in_tool:
36
                print(" 完成", flush=True)
37
                in_tool = False
38

39
        elif isinstance(message, ResultMessage):
40
            print(f"\n\n── 任务结束（{message.subtype}），花费 ${message.total_cost_usd:.4f} ──")
41

42

43
asyncio.run(stream_with_progress())

1
import { query } from "@anthropic-ai/claude-agent-sdk";
2

3
let inTool = false;
4

5
for await (const message of query({
6
  prompt: "在代码库里找出所有 TODO 注释，按文件汇总。",
7
  options: {
8
    includePartialMessages: true,
9
    allowedTools: ["Read", "Glob", "Grep"],
10
    maxTurns: 15,
11
  },
12
})) {
13
  if (message.type === "stream_event") {
14
    const ev = message.event;
15

16
    if (ev.type === "content_block_start" && ev.content_block.type === "tool_use") {
17
      process.stdout.write(`\n⟳ ${ev.content_block.name}...`);
18
      inTool = true;
19
    } else if (ev.type === "content_block_delta") {
20
      if (ev.delta.type === "text_delta" && !inTool) {
21
        process.stdout.write(ev.delta.text);
22
      }
23
    } else if (ev.type === "content_block_stop" && inTool) {
24
      console.log(" 完成");
25
      inTool = false;
26
    }
27
  } else if (message.type === "result") {
28
    console.log(`\n\n── 任务结束（${message.subtype}），花费 $${message.total_cost_usd.toFixed(4)} ──`);
29
  }
30
}

1
import asyncio
2
from claude_agent_sdk import query, ClaudeAgentOptions, SystemMessage, ResultMessage
3

4

5
async def multi_turn():
6
    session_id = None
7

8
    # 第一轮：读取项目结构
9
    async for message in query(
10
        prompt="读取项目的认证模块，理解它的结构。",
11
        options=ClaudeAgentOptions(allowed_tools=["Read", "Glob"]),
12
    ):
13
        if isinstance(message, SystemMessage) and message.subtype == "init":
14
            session_id = message.data["session_id"]
15
        elif isinstance(message, ResultMessage) and message.subtype == "success":
16
            print("第一轮完成：", message.result[:100], "...")
17

18
    print(f"\n会话 ID：{session_id}")
19

20
    # 第二轮：在同一会话上下文中继续
21
    async for message in query(
22
        prompt="现在找出所有调用了认证模块的地方。",  # 'it' 指代上轮读取的模块
23
        options=ClaudeAgentOptions(
24
            resume=session_id,
25
            allowed_tools=["Glob", "Grep"],
26
        ),
27
    ):
28
        if isinstance(message, ResultMessage) and message.subtype == "success":
29
            print("\n第二轮完成：", message.result[:200])
30

31

32
asyncio.run(multi_turn())

1
options = ClaudeAgentOptions(cwd="/work/tenant-a-session-xyz")

1
options: { cwd: "/work/tenant-a-session-xyz" }

1
可承载代理数 = (主机 RAM - 系统开销) / 每个会话峰值 RSS

1
options = ClaudeAgentOptions(
2
    cwd=tenant_dir,
3
    setting_sources=[],       # 不加载文件系统设置（CLAUDE.md 等）
4
    env={
5
        "CLAUDE_CONFIG_DIR": f"/configs/{tenant_id}",
6
        "CLAUDE_CODE_DISABLE_AUTO_MEMORY": "1",  # 禁用自动内存注入
7
    },
8
)

1
// TypeScript 的 env 是替换而非合并，必须展开 process.env
2
options: {
3
  cwd: tenantDir,
4
  settingSources: [],
5
  env: {
6
    ...process.env,                          // 保留 PATH、ANTHROPIC_API_KEY 等
7
    CLAUDE_CONFIG_DIR: `/configs/${tenantId}`,
8
    CLAUDE_CODE_DISABLE_AUTO_MEMORY: "1",
9
  },
10
}