Claude Agent SDK 完全指南 - 从入门到精通

什么是 Claude Agent SDK?

Claude Agent SDK 是 Anthropic 官方提供的一套开发工具包，让开发者能够以编程方式构建 AI 代理（Agent）。简单来说，它就像是把 Claude Code 的能力打包成了一个可以在代码中调用的库。

类比理解

想象你有一个超级助手，这个助手不仅能理解你的指令，还能：

读取文件：像你一样打开并阅读项目中的任何文件
编辑代码：精确地修改代码中的某一行或某一段
执行命令：在终端中运行 npm install、git commit 等命令
搜索网络：查找最新的文档或解决方案

而 Claude Agent SDK 就是把这个超级助手变成了你可以在代码中调用的 API。

与普通 API 的区别

特性	普通 Claude API	Claude Agent SDK
工具执行	需要自己实现	内置自动执行
文件操作	不支持	原生支持
命令执行	不支持	原生支持
上下文管理	手动管理	自动管理
会话恢复	需要自己实现	内置支持

快速开始

安装

首先需要安装 Claude Code（SDK 使用它作为运行时）：

# macOS/Linux/WSL
curl -fsSL https://claude.ai/install.sh | bash

# Homebrew
brew install --cask claude-code

# Windows (WinGet)
winget install Anthropic.ClaudeCode

然后安装 SDK：

# TypeScript
npm install @anthropic-ai/claude-agent-sdk

# Python (推荐使用 uv)
uv init && uv add claude-agent-sdk

# Python (pip)
pip install claude-agent-sdk

设置 API Key

export ANTHROPIC_API_KEY=your-api-key

如果你已经通过 claude 命令行登录过，SDK 会自动使用该认证信息。

第一个 Agent

TypeScript 版本：

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

// 创建一个能读取和搜索文件的 Agent
for await (const message of query({
  prompt: "分析当前目录下的所有 TypeScript 文件，找出潜在的性能问题",
  options: {
    allowedTools: ["Read", "Glob", "Grep"],
    permissionMode: "bypassPermissions"  // 自动批准所有操作
  }
})) {
  if ("result" in message) {
    console.log(message.result);
  }
}

Python 版本：

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions

async def main():
    # 创建一个能读取和搜索文件的 Agent
    async for message in query(
        prompt="分析当前目录下的所有 TypeScript 文件，找出潜在的性能问题",
        options=ClaudeAgentOptions(
            allowed_tools=["Read", "Glob", "Grep"],
            permission_mode="bypassPermissions"  # 自动批准所有操作
        )
    ):
        if hasattr(message, "result"):
            print(message.result)

asyncio.run(main())

核心概念

1. 内置工具 (Built-in Tools)

SDK 提供了一系列开箱即用的工具，无需你自己实现：

工具	功能	使用场景
`Read`	读取文件内容	分析代码、读取配置
`Write`	创建新文件	生成代码、创建配置文件
`Edit`	精确编辑现有文件	修复 Bug、重构代码
`Bash`	执行终端命令	运行测试、Git 操作、安装依赖
`Glob`	按模式查找文件	查找所有 `.ts` 文件
`Grep`	搜索文件内容	查找特定函数或变量的使用
`WebSearch`	搜索网络	查找最新文档和解决方案
`WebFetch`	获取网页内容	抓取 API 文档
`Task`	启动子代理	并行处理复杂任务

工具组合示例

TypeScript - 只读分析 Agent：

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

// 只读分析 Agent - 安全地分析代码但不做任何修改
for await (const message of query({
  prompt: "检查所有 TODO 注释并生成任务清单",
  options: {
    allowedTools: ["Read", "Glob", "Grep"],  // 只有读取权限
    permissionMode: "bypassPermissions"
  }
})) {
  if (message.type === "assistant") {
    for (const block of message.message?.content || []) {
      if ("text" in block) {
        console.log(block.text);
      }
    }
  }
}

Python - 只读分析 Agent：

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, TextBlock

async def main():
    # 只读分析 Agent - 安全地分析代码但不做任何修改
    async for message in query(
        prompt="检查所有 TODO 注释并生成任务清单",
        options=ClaudeAgentOptions(
            allowed_tools=["Read", "Glob", "Grep"],  # 只有读取权限
            permission_mode="bypassPermissions"
        )
    ):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(block.text)

asyncio.run(main())

2. 权限模式 (Permission Modes)

权限模式决定了 Agent 在执行工具时的行为：

flowchart TD
    A[工具调用请求] --> B{检查 Hooks}
    B -->|阻止| C[操作被拒绝]
    B -->|允许/继续| D{检查权限规则}
    D -->|deny 规则匹配| C
    D -->|allow 规则匹配| E[执行工具]
    D -->|ask 规则匹配| F{canUseTool 回调}
    D -->|无匹配| G{检查权限模式}
    G -->|bypassPermissions| E
    G -->|acceptEdits| H{是文件操作?}
    H -->|是| E
    H -->|否| F
    G -->|default| F
    G -->|plan| C
    F -->|批准| E
    F -->|拒绝| C

模式	描述	适用场景
`default`	默认模式，未匹配的工具触发回调	需要人工审批的场景
`acceptEdits`	自动批准文件编辑操作	信任 Agent 的代码修改
`bypassPermissions`	跳过所有权限检查	CI/CD、完全信任的环境
`plan`	规划模式，不执行任何工具	代码审查、方案设计

动态切换权限模式

TypeScript 版本：

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

async function main() {
  const q = query({
    prompt: "重构 utils.py 中的代码",
    options: {
      permissionMode: "default",  // 开始时使用默认模式
    },
  });

  // 在确认 Agent 的计划后，切换到自动批准模式
  await q.setPermissionMode("acceptEdits");

  for await (const message of q) {
    if ("result" in message) {
      console.log(message.result);
    }
  }
}

main();

Python 版本：

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions

async def main():
    q = query(
        prompt="重构 utils.py 中的代码",
        options=ClaudeAgentOptions(
            permission_mode="default",  # 开始时使用默认模式
        ),
    )

    # 在确认 Agent 的计划后，切换到自动批准模式
    await q.set_permission_mode("acceptEdits")

    async for message in q:
        if hasattr(message, "result"):
            print(message.result)

asyncio.run(main())

3. 钩子系统 (Hooks)

Hooks 让你能在 Agent 执行的关键节点插入自定义逻辑，就像 Git Hooks 或 React Hooks 一样。

实战：安全审计 Hook

TypeScript 版本：

import { query, HookCallback, PreToolUseHookInput } from "@anthropic-ai/claude-agent-sdk";
import { appendFileSync } from "fs";

// 定义一个审计日志 Hook
const auditLogger: HookCallback = async (input, toolUseID, { signal }) => {
  const preInput = input as PreToolUseHookInput;

  // 记录所有工具调用到审计日志
  const logEntry = {
    timestamp: new Date().toISOString(),
    tool: preInput.tool_name,
    input: preInput.tool_input,
    toolUseId: toolUseID
  };

  appendFileSync("./audit.log", JSON.stringify(logEntry) + "\n");

  return {};  // 返回空对象表示允许继续
};

// 定义一个安全检查 Hook - 阻止危险命令
const securityGuard: HookCallback = async (input, toolUseID, { signal }) => {
  const preInput = input as PreToolUseHookInput;

  if (preInput.tool_name === "Bash") {
    const command = preInput.tool_input?.command as string;
    const dangerousPatterns = [
      "rm -rf /",
      "rm -rf ~",
      ":(){ :|:& };:",  // Fork bomb
      "> /dev/sda",
    ];

    for (const pattern of dangerousPatterns) {
      if (command?.includes(pattern)) {
        return {
          hookSpecificOutput: {
            hookEventName: input.hook_event_name,
            permissionDecision: "deny",
            permissionDecisionReason: `检测到危险命令: ${pattern}`
          }
        };
      }
    }
  }

  return {};
};

// 使用 Hooks
for await (const message of query({
  prompt: "清理项目中的临时文件",
  options: {
    allowedTools: ["Bash", "Glob", "Read"],
    hooks: {
      PreToolUse: [
        { hooks: [auditLogger] },  // 所有工具都记录日志
        { matcher: "Bash", hooks: [securityGuard] }  // 只对 Bash 检查安全
      ]
    }
  }
})) {
  if ("result" in message) {
    console.log(message.result);
  }
}

Python 版本：

import asyncio
import json
from datetime import datetime
from claude_agent_sdk import query, ClaudeAgentOptions, HookMatcher

# 定义一个审计日志 Hook
async def audit_logger(input_data, tool_use_id, context):
    # 记录所有工具调用到审计日志
    log_entry = {
        "timestamp": datetime.now().isoformat(),
        "tool": input_data["tool_name"],
        "input": input_data["tool_input"],
        "tool_use_id": tool_use_id
    }

    with open("./audit.log", "a") as f:
        f.write(json.dumps(log_entry) + "\n")

    return {}  # 返回空对象表示允许继续


# 定义一个安全检查 Hook - 阻止危险命令
async def security_guard(input_data, tool_use_id, context):
    if input_data["tool_name"] == "Bash":
        command = input_data["tool_input"].get("command", "")
        dangerous_patterns = [
            "rm -rf /",
            "rm -rf ~",
            ":(){ :|:& };:",  # Fork bomb
            "> /dev/sda",
        ]

        for pattern in dangerous_patterns:
            if pattern in command:
                return {
                    "hookSpecificOutput": {
                        "hookEventName": input_data["hook_event_name"],
                        "permissionDecision": "deny",
                        "permissionDecisionReason": f"检测到危险命令: {pattern}"
                    }
                }

    return {}


async def main():
    # 使用 Hooks
    async for message in query(
        prompt="清理项目中的临时文件",
        options=ClaudeAgentOptions(
            allowed_tools=["Bash", "Glob", "Read"],
            hooks={
                "PreToolUse": [
                    HookMatcher(hooks=[audit_logger]),  # 所有工具都记录日志
                    HookMatcher(matcher="Bash", hooks=[security_guard])  # 只对 Bash 检查安全
                ]
            }
        )
    ):
        if hasattr(message, "result"):
            print(message.result)

asyncio.run(main())

4. 自定义工具 (Custom Tools)

你可以通过 MCP（Model Context Protocol）创建自定义工具，让 Agent 能够调用你自己的 API 或服务。

类比理解

把自定义工具想象成给 Agent 安装新的”技能”：

Agent 本身会读写文件、执行命令
但如果你想让它查询数据库、调用支付 API，就需要自定义工具
自定义工具就像是给 Agent 开发的”插件”

TypeScript 版本：

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

// 创建一个天气查询工具
const weatherTool = tool(
  "get_weather",
  "获取指定城市的当前天气",
  {
    city: z.string().describe("城市名称，如：北京、上海"),
    unit: z.enum(["celsius", "fahrenheit"]).default("celsius").describe("温度单位")
  },
  async (args) => {
    // 这里调用真实的天气 API
    const response = await fetch(
      `https://api.weatherapi.com/v1/current.json?key=${process.env.WEATHER_API_KEY}&q=${args.city}`
    );
    const data = await response.json();

    const temp = args.unit === "celsius"
      ? data.current.temp_c
      : data.current.temp_f;

    return {
      content: [{
        type: "text",
        text: `${args.city}的当前温度是 ${temp}${args.unit === "celsius" ? "°C" : "°F"}，天气状况：${data.current.condition.text}`
      }]
    };
  }
);

// 创建一个数据库查询工具
const dbQueryTool = tool(
  "query_database",
  "执行 SQL 查询并返回结果",
  {
    query: z.string().describe("SQL 查询语句"),
    database: z.enum(["users", "orders", "products"]).describe("目标数据库")
  },
  async (args) => {
    // 这里连接真实的数据库
    // const results = await db.query(args.query);

    // 示例返回
    return {
      content: [{
        type: "text",
        text: `查询 ${args.database} 数据库完成，返回 10 条记录`
      }]
    };
  }
);

// 创建 SDK MCP 服务器
const customServer = createSdkMcpServer({
  name: "my-tools",
  version: "1.0.0",
  tools: [weatherTool, dbQueryTool]
});

// 使用自定义工具的消息生成器
async function* generateMessages() {
  yield {
    type: "user" as const,
    message: {
      role: "user" as const,
      content: "查询北京的天气，然后统计今天的订单数量"
    }
  };
}

// 运行 Agent
for await (const message of query({
  prompt: generateMessages(),
  options: {
    mcpServers: {
      "my-tools": customServer
    },
    allowedTools: [
      "mcp__my-tools__get_weather",
      "mcp__my-tools__query_database"
    ]
  }
})) {
  if ("result" in message) {
    console.log(message.result);
  }
}

Python 版本：

import asyncio
from typing import Any
import aiohttp
import os
from claude_agent_sdk import (
    query,
    ClaudeAgentOptions,
    tool,
    create_sdk_mcp_server,
    ClaudeSDKClient
)


# 创建一个天气查询工具
@tool(
    "get_weather",
    "获取指定城市的当前天气",
    {"city": str, "unit": str}
)
async def get_weather(args: dict[str, Any]) -> dict[str, Any]:
    """获取天气信息"""
    city = args["city"]
    unit = args.get("unit", "celsius")

    # 这里调用真实的天气 API
    async with aiohttp.ClientSession() as session:
        async with session.get(
            f"https://api.weatherapi.com/v1/current.json",
            params={"key": os.environ["WEATHER_API_KEY"], "q": city}
        ) as response:
            data = await response.json()

    temp = data["current"]["temp_c"] if unit == "celsius" else data["current"]["temp_f"]
    unit_symbol = "°C" if unit == "celsius" else "°F"

    return {
        "content": [{
            "type": "text",
            "text": f"{city}的当前温度是 {temp}{unit_symbol}，天气状况：{data['current']['condition']['text']}"
        }]
    }


# 创建一个数据库查询工具
@tool(
    "query_database",
    "执行 SQL 查询并返回结果",
    {"query": str, "database": str}
)
async def query_database(args: dict[str, Any]) -> dict[str, Any]:
    """执行数据库查询"""
    # 这里连接真实的数据库
    # results = await db.query(args["query"])

    # 示例返回
    return {
        "content": [{
            "type": "text",
            "text": f"查询 {args['database']} 数据库完成，返回 10 条记录"
        }]
    }


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


async def main():
    # 使用自定义工具
    options = ClaudeAgentOptions(
        mcp_servers={"my-tools": custom_server},
        allowed_tools=[
            "mcp__my-tools__get_weather",
            "mcp__my-tools__query_database"
        ]
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("查询北京的天气，然后统计今天的订单数量")

        async for msg in client.receive_response():
            print(msg)

asyncio.run(main())

5. MCP 服务器集成

MCP（Model Context Protocol）是一个开放标准，让 Agent 能够连接到外部工具和数据源。

flowchart LR
    A[Claude Agent] --> B{MCP 协议}
    B --> C[Stdio 本地进程]
    B --> D[HTTP/SSE 远程服务]
    B --> E[SDK 内置服务器]

    C --> F[GitHub MCP]
    C --> G[Postgres MCP]
    C --> H[Filesystem MCP]

    D --> I[云端 API]
    D --> J[第三方服务]

    E --> K[自定义工具]

连接 GitHub MCP 服务器

TypeScript 版本：

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

// 连接 GitHub MCP 服务器
for await (const message of query({
  prompt: "列出 anthropics/claude-code 仓库的最近 5 个 issue",
  options: {
    mcpServers: {
      "github": {
        command: "npx",
        args: ["-y", "@modelcontextprotocol/server-github"],
        env: {
          GITHUB_TOKEN: process.env.GITHUB_TOKEN
        }
      }
    },
    allowedTools: ["mcp__github__list_issues", "mcp__github__search_issues"]
  }
})) {
  // 验证 MCP 服务器连接
  if (message.type === "system" && message.subtype === "init") {
    console.log("MCP 服务器:", message.mcp_servers);
  }

  if ("result" in message) {
    console.log(message.result);
  }
}

Python 版本：

import asyncio
import os
from claude_agent_sdk import (
    query,
    ClaudeAgentOptions,
    ResultMessage,
    SystemMessage
)

async def main():
    options = ClaudeAgentOptions(
        mcp_servers={
            "github": {
                "command": "npx",
                "args": ["-y", "@modelcontextprotocol/server-github"],
                "env": {
                    "GITHUB_TOKEN": os.environ["GITHUB_TOKEN"]
                }
            }
        },
        allowed_tools=["mcp__github__list_issues", "mcp__github__search_issues"]
    )

    async for message in query(
        prompt="列出 anthropics/claude-code 仓库的最近 5 个 issue",
        options=options
    ):
        # 验证 MCP 服务器连接
        if isinstance(message, SystemMessage) and message.subtype == "init":
            print("MCP 服务器:", message.data.get("mcp_servers"))

        if isinstance(message, ResultMessage) and message.subtype == "success":
            print(message.result)

asyncio.run(main())

使用配置文件（推荐）

在项目根目录创建 .mcp.json：

{
  "mcpServers": {
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {
        "GITHUB_TOKEN": "${GITHUB_TOKEN}"
      }
    },
    "postgres": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-postgres", "${DATABASE_URL}"]
    },
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "/Users/me/projects"]
    }
  }
}

SDK 会自动加载这个配置文件。

6. 会话管理 (Sessions)

会话管理让你能够跨多次交互保持上下文，就像 Git 的分支一样，你可以继续、分叉会话。

类比理解

把会话想象成一个”存档点”：

你和 Agent 的对话有了一个存档
之后可以从这个存档继续
也可以从同一个存档开始，走向不同的方向（分叉）

TypeScript 版本：

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

let sessionId: string | undefined;

// 第一次对话 - 获取 session ID
for await (const message of query({
  prompt: "帮我设计一个 REST API",
  options: { model: "claude-sonnet-4-5" }
})) {
  if (message.type === "system" && message.subtype === "init") {
    sessionId = message.session_id;
    console.log(`会话已创建: ${sessionId}`);
  }
}

// 继续同一个会话
if (sessionId) {
  for await (const message of query({
    prompt: "现在给这个 API 添加认证功能",
    options: {
      resume: sessionId,  // 恢复会话
      model: "claude-sonnet-4-5"
    }
  })) {
    if ("result" in message) {
      console.log(message.result);
    }
  }
}

// 分叉会话 - 从同一个点尝试不同方案
if (sessionId) {
  for await (const message of query({
    prompt: "改成设计 GraphQL API",
    options: {
      resume: sessionId,
      forkSession: true,  // 创建新分支
      model: "claude-sonnet-4-5"
    }
  })) {
    if (message.type === "system" && message.subtype === "init") {
      console.log(`分叉会话: ${message.session_id}`);  // 新的 session ID
    }
  }
}

Python 版本：

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions

async def main():
    session_id = None

    # 第一次对话 - 获取 session ID
    async for message in query(
        prompt="帮我设计一个 REST API",
        options=ClaudeAgentOptions(model="claude-sonnet-4-5")
    ):
        if hasattr(message, "subtype") and message.subtype == "init":
            session_id = message.data.get("session_id")
            print(f"会话已创建: {session_id}")

    # 继续同一个会话
    if session_id:
        async for message in query(
            prompt="现在给这个 API 添加认证功能",
            options=ClaudeAgentOptions(
                resume=session_id,  # 恢复会话
                model="claude-sonnet-4-5"
            )
        ):
            if hasattr(message, "result"):
                print(message.result)

    # 分叉会话 - 从同一个点尝试不同方案
    if session_id:
        async for message in query(
            prompt="改成设计 GraphQL API",
            options=ClaudeAgentOptions(
                resume=session_id,
                fork_session=True,  # 创建新分支
                model="claude-sonnet-4-5"
            )
        ):
            if hasattr(message, "subtype") and message.subtype == "init":
                forked_id = message.data.get("session_id")
                print(f"分叉会话: {forked_id}")  # 新的 session ID

asyncio.run(main())

7. 子代理 (Subagents)

子代理是独立的 Agent 实例，主 Agent 可以派遣它们处理特定子任务。

flowchart TB
    A[主 Agent] --> B[Task 工具]
    B --> C[code-reviewer 子代理]
    B --> D[test-runner 子代理]
    B --> E[doc-writer 子代理]

    C --> F[返回代码审查结果]
    D --> G[返回测试结果]
    E --> H[返回文档]

    F --> A
    G --> A
    H --> A

子代理的优势

优势	说明	示例
上下文隔离	子代理有独立上下文，不污染主对话	研究助手探索数十个文件，只返回关键发现
并行执行	多个子代理可以同时运行	同时运行代码检查、安全扫描、测试覆盖
专业化	每个子代理可以有专门的指令和知识	数据库迁移专家具备 SQL 最佳实践
工具限制	可以限制子代理只使用特定工具	文档审查器只能读取不能修改

TypeScript 版本：

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

for await (const message of query({
  prompt: "审查认证模块的安全问题",
  options: {
    // Task 工具是必须的，子代理通过它被调用
    allowedTools: ["Read", "Glob", "Grep", "Task"],
    agents: {
      "code-reviewer": {
        // description 告诉 Claude 何时使用这个子代理
        description: "代码审查专家，用于质量、安全和可维护性审查",
        // prompt 定义子代理的行为和专业知识
        prompt: `你是一个代码审查专家，专注于安全、性能和最佳实践。

审查代码时请：
- 识别安全漏洞
- 检查性能问题
- 验证代码规范遵守情况
- 提出具体改进建议

反馈要详尽但简洁。`,
        // tools 限制子代理可用的工具（这里是只读）
        tools: ["Read", "Grep", "Glob"],
        // model 可以为子代理指定不同的模型
        model: "sonnet"
      },
      "test-runner": {
        description: "运行和分析测试套件，用于测试执行和覆盖率分析",
        prompt: `你是测试执行专家。运行测试并清晰分析结果。

关注：
- 执行测试命令
- 分析测试输出
- 识别失败的测试
- 建议修复方案`,
        // Bash 权限让这个子代理可以运行测试
        tools: ["Bash", "Read", "Grep"]
      }
    }
  }
})) {
  if ("result" in message) {
    console.log(message.result);
  }
}

Python 版本：

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, AgentDefinition

async def main():
    async for message in query(
        prompt="审查认证模块的安全问题",
        options=ClaudeAgentOptions(
            # Task 工具是必须的，子代理通过它被调用
            allowed_tools=["Read", "Glob", "Grep", "Task"],
            agents={
                "code-reviewer": AgentDefinition(
                    # description 告诉 Claude 何时使用这个子代理
                    description="代码审查专家，用于质量、安全和可维护性审查",
                    # prompt 定义子代理的行为和专业知识
                    prompt="""你是一个代码审查专家，专注于安全、性能和最佳实践。

审查代码时请：
- 识别安全漏洞
- 检查性能问题
- 验证代码规范遵守情况
- 提出具体改进建议

反馈要详尽但简洁。""",
                    # tools 限制子代理可用的工具（这里是只读）
                    tools=["Read", "Grep", "Glob"],
                    # model 可以为子代理指定不同的模型
                    model="sonnet"
                ),
                "test-runner": AgentDefinition(
                    description="运行和分析测试套件，用于测试执行和覆盖率分析",
                    prompt="""你是测试执行专家。运行测试并清晰分析结果。

关注：
- 执行测试命令
- 分析测试输出
- 识别失败的测试
- 建议修复方案""",
                    # Bash 权限让这个子代理可以运行测试
                    tools=["Bash", "Read", "Grep"]
                )
            }
        )
    ):
        if hasattr(message, "result"):
            print(message.result)

asyncio.run(main())

高级用法

流式输出 vs 单次输出

模式	适用场景	特点
流式输出	交互式应用、实时显示	可以看到 Agent 的思考过程
单次输出	CI/CD、批处理	等待完成后一次性获取结果

TypeScript - 流式 vs 单次输出：

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

// 流式输出 - 实时显示进度
console.log("=== 流式输出 ===");
for await (const message of query({
  prompt: "分析这个项目的架构",
  options: { allowedTools: ["Read", "Glob", "Grep"] }
})) {
  if (message.type === "assistant" && message.message?.content) {
    for (const block of message.message.content) {
      if ("text" in block) {
        process.stdout.write(block.text);  // 实时输出
      }
    }
  }
}

// 单次输出 - 收集所有消息
console.log("\n=== 单次输出 ===");
const messages = [];
for await (const message of query({
  prompt: "分析这个项目的架构",
  options: { allowedTools: ["Read", "Glob", "Grep"] }
})) {
  messages.push(message);
}

// 处理收集的消息
const result = messages.find(m => "result" in m);
if (result && "result" in result) {
  console.log(result.result);
}

Python - 流式 vs 单次输出：

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, TextBlock

async def main():
    # 流式输出 - 实时显示进度
    print("=== 流式输出 ===")
    async for message in query(
        prompt="分析这个项目的架构",
        options=ClaudeAgentOptions(allowed_tools=["Read", "Glob", "Grep"])
    ):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(block.text, end="", flush=True)  # 实时输出

    # 单次输出 - 收集所有消息
    print("\n=== 单次输出 ===")
    messages = []
    async for message in query(
        prompt="分析这个项目的架构",
        options=ClaudeAgentOptions(allowed_tools=["Read", "Glob", "Grep"])
    ):
        messages.append(message)

    # 处理收集的消息
    for msg in messages:
        if hasattr(msg, "result"):
            print(msg.result)
            break

asyncio.run(main())

错误处理

TypeScript 版本：

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

try {
  for await (const message of query({
    prompt: "分析代码",
    options: { allowedTools: ["Read"] }
  })) {
    // 检查 MCP 服务器连接状态
    if (message.type === "system" && message.subtype === "init") {
      const failedServers = message.mcp_servers?.filter(
        s => s.status !== "connected"
      );

      if (failedServers && failedServers.length > 0) {
        console.warn("MCP 连接失败:", failedServers);
      }
    }

    // 检查执行错误
    if (message.type === "result" && message.subtype === "error_during_execution") {
      console.error("执行过程中出错");
    }

    if ("result" in message) {
      console.log(message.result);
    }
  }
} catch (error) {
  if (error instanceof Error) {
    if (error.name === "CLINotFoundError") {
      console.error("请安装 Claude Code: curl -fsSL https://claude.ai/install.sh | bash");
    } else if (error.name === "ProcessError") {
      console.error("进程执行失败:", error.message);
    } else {
      console.error("未知错误:", error.message);
    }
  }
}

Python 版本：

import asyncio
from claude_agent_sdk import (
    query,
    ClaudeAgentOptions,
    ClaudeSDKError,
    CLINotFoundError,
    CLIConnectionError,
    ProcessError,
    CLIJSONDecodeError,
    SystemMessage,
    ResultMessage
)

async def main():
    try:
        async for message in query(
            prompt="分析代码",
            options=ClaudeAgentOptions(allowed_tools=["Read"])
        ):
            # 检查 MCP 服务器连接状态
            if isinstance(message, SystemMessage) and message.subtype == "init":
                mcp_servers = message.data.get("mcp_servers", [])
                failed_servers = [
                    s for s in mcp_servers
                    if s.get("status") != "connected"
                ]

                if failed_servers:
                    print(f"MCP 连接失败: {failed_servers}")

            # 检查执行错误
            if isinstance(message, ResultMessage):
                if message.subtype == "error_during_execution":
                    print("执行过程中出错")

            if hasattr(message, "result"):
                print(message.result)

    except CLINotFoundError:
        print("请安装 Claude Code: curl -fsSL https://claude.ai/install.sh | bash")
    except ProcessError as e:
        print(f"进程执行失败，退出码: {e.exit_code}")
    except CLIJSONDecodeError as e:
        print(f"JSON 解析失败: {e}")
    except ClaudeSDKError as e:
        print(f"SDK 错误: {e}")

asyncio.run(main())

生产环境部署

Docker 部署

FROM node:18-slim

# 安装 Claude Code
RUN curl -fsSL https://claude.ai/install.sh | bash

# 设置工作目录
WORKDIR /app

# 复制项目文件
COPY package*.json ./
RUN npm install

COPY . .

# 设置环境变量
ENV ANTHROPIC_API_KEY=""

# 运行应用
CMD ["node", "agent.js"]

最佳实践

API 参考

ClaudeAgentOptions

参数	类型	说明
`allowed_tools` / `allowedTools`	`string[]`	允许使用的工具列表
`permission_mode` / `permissionMode`	`string`	权限模式
`system_prompt` / `systemPrompt`	`string`	自定义系统提示
`max_turns` / `maxTurns`	`number`	最大轮数
`model`	`string`	使用的模型
`mcp_servers` / `mcpServers`	`object`	MCP 服务器配置
`hooks`	`object`	钩子配置
`agents`	`object`	子代理定义
`cwd`	`string`	工作目录
`resume`	`string`	恢复会话 ID
`fork_session` / `forkSession`	`boolean`	是否分叉会话

消息类型

类型	说明
`SystemMessage`	系统消息，包含初始化信息
`AssistantMessage`	Claude 的响应，包含文本和工具调用
`UserMessage`	用户消息
`ResultMessage`	最终结果消息

常见问题

Q: Claude Code not found 错误怎么办？

安装 Claude Code 并重启终端：

curl -fsSL https://claude.ai/install.sh | bash

Q: API key not found 错误怎么办？

设置环境变量：

export ANTHROPIC_API_KEY=your-api-key

或在代码中使用 .env 文件。

Q: 如何减少 Token 使用？

使用 max_turns 限制轮数
为简单任务使用 haiku 模型
启用 MCP Tool Search（自动按需加载工具）
使用子代理隔离上下文

Q: 如何确保安全？

使用 PreToolUse Hook 审计和阻止危险操作
限制 allowed_tools 只包含必要工具
避免在不可信环境使用 bypassPermissions
使用沙箱目录限制文件访问范围

什么是 Claude Agent SDK?

类比理解

与普通 API 的区别

快速开始

安装

设置 API Key

第一个 Agent

核心概念

1. 内置工具 (Built-in Tools)

工具组合示例

2. 权限模式 (Permission Modes)

动态切换权限模式

3. 钩子系统 (Hooks)

实战：安全审计 Hook

4. 自定义工具 (Custom Tools)

类比理解

5. MCP 服务器集成

连接 GitHub MCP 服务器

使用配置文件（推荐）

6. 会话管理 (Sessions)

类比理解

7. 子代理 (Subagents)

子代理的优势

高级用法

流式输出 vs 单次输出

错误处理

生产环境部署

Docker 部署

最佳实践

API 参考

ClaudeAgentOptions

消息类型

常见问题

Q: Claude Code not found 错误怎么办？

Q: API key not found 错误怎么办？

Q: 如何减少 Token 使用？

Q: 如何确保安全？

参考资源

Read Next

SwiftUI 基础 - 给 React 开发者的入门指南

Read Previous

Vercel AI SDK Agent Loop 完全指南