Claude Code(十)Claude Code SDK
你可以用 Python 或 TypeScript 编写代码像调用普通函数一样调用 AI Agent。在自己的应用里调用 Claude Code 的能力。现在你需要的不是一个命令行工具而是一个可编程的 SDK。这就是 Claude Agent SDK 的价值——它把 Claude Code 的所有能力封装成了可编程的接口。你可以用 Python 或 TypeScript 编写代码像调用普通函数一样调用 AI Agent。一、Claude Agent SDKClaude Agent SDK 提供了可编程的 Claude Code。它不是一个新的模型 API而是对 Claude Code 这个 Agent 系统的完整封装。你通过 SDK 调用的不是一个简单的文本生成接口而是一个完整的 Agent 循环——Claude 会自主决定使用哪些工具、读取哪些文件、执行哪些命令然后把结果返回给你。Agent SDK 支持两种语言。Pythonpip install claude-agent-sdkTypeScriptnpm install anthropic-ai/claude-agent-sdk两种语言的 API 设计保持一致功能完全相同。SDK 能力一览下面这张表列出了 Agent SDK 赋予你的全部能力。每一项都对应 Claude Code 本身的一种工具SDK 让你可以在自己的代码中精确控制这些工具的使用。用户上传代码后Agent 可以用 Read 读取文件、用 Grep 搜索模式、用 Glob 遍历目录、用 Bash 运行测试——所有这些操作都在你的应用后端自动完成用户只需要等待报告生成。两种使用方式Agent SDK 提供了两种使用方式适用于不同场景。理解它们的区别是正确使用 SDK 的第一步。你可以把它们类比为 Python 中的 requests.get() 和 requests.Session()前者是无状态的一次性调用后者是有状态的会话管理。query() 函数简洁高效query() 是最简单的方式适合轻量级用例。它接收一个 Prompt 字符串返回一个异步迭代器你可以逐条接收 Agent 产生的消息。整个过程不需要手动管理连接、配置选项或处理会话状态SDK 帮你搞定一切。这种设计的好处是显而易见的当你只想快速验证一个想法、写一个脚本、或者做一次性的分析时不需要写二十行初始化代码。一个函数调用就够了。from claude_agent_sdk import query import asyncio async def main(): # 简单查询 async for message in query(解释什么是递归): if message.type text: print(message.text) asyncio.run(main())ClaudeSDKClient 类完整控制当你需要更精细的控制时比如限制 Agent 只能使用特定工具、设置最大执行轮次、管理多轮会话就需要使用 ClaudeSDKClient。它提供了完整的配置能力让你可以像搭积木一样组合 Agent 的行为。与开箱即用的query() 不同ClaudeSDKClient 要求你显式地创建客户端、配置选项、管理连接生命周期。这种显式性是刻意为之的在生产环境中你需要明确知道 Agent 能做什么、不能做什么、在什么条件下停止。隐式的默认值在生产中往往是 Bug 的温床。from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions import asyncio async def main(): options ClaudeAgentOptions( allowed_tools[Read, Grep, Glob], max_turns10, permission_modeplan # 只读模式 ) async with ClaudeSDKClient(optionsoptions) as client: await client.query(分析 src/ 目录的代码结构) async for message in client.receive_response(): if message.type text: print(message.text) elif message.type tool_use: print(fUsing tool: {message.tool_name}) asyncio.run(main())ClaudeAgentOptions 配置详解ClaudeAgentOptions 是控制 Agent 行为的核心配置类。你可以把它理解为 Agent 的“说明书”它告诉 Agent 该用什么模型、能用什么工具、最多跑几轮、在什么目录下工作。每一个配置项都会直接影响 Agent 的行为和成本。下面是完整的配置项。不需要一次记住所有配置你可以先关注最常用的四个allowed_tools、permission_mode、max_turns、model。其余的在需要时查阅即可。from claude_agent_sdk import ClaudeAgentOptions options ClaudeAgentOptions( # 模型选择 modelsonnet, # sonnet | opus | haiku # 工具控制 allowed_tools[Read, Write, Bash, Grep, Glob], disallowed_tools[Task], # 权限模式 permission_modedefault, # default | acceptEdits | plan | bypass # 执行控制 max_turns20, cwd/path/to/project, # 输出格式 output_formatstream-json, # text | json | stream-json # 会话管理 continue_conversationTrue, resumesession-id, # 系统提示 system_promptYou are a helpful coding assistant., # MCP 服务器 mcp_servers{ my-server: {...} }, # Hooks hooks{ PreToolUse: [...], PostToolUse: [...] } )权限模式详解# 代码审查场景只读 options ClaudeAgentOptions( permission_modeplan, allowed_tools[Read, Grep, Glob] ) # 自动修复场景接受编辑 options ClaudeAgentOptions( permission_modeacceptEdits, allowed_tools[Read, Write, Edit] )内置工具列表消息类型与响应处理async for message in client.receive_response(): match message.type: case text: # 文本响应 print(message.text) case tool_use: # 工具调用Agent 正在使用工具 print(fTool: {message.tool_name}) print(fInput: {message.tool_input}) case tool_result: # 工具执行结果 print(fResult: {message.result}) case error: # 错误信息 print(fError: {message.error}) case result: # 最终结果任务完成 print(fFinal: {message.result}) print(fCost: ${message.total_cost_usd})会话管理实战项目——代码分析 Agent#!/usr/bin/env python3 代码分析 Agent 使用 Claude Agent SDK 构建一个自动代码分析工具。 import asyncio import sys from datetime import datetime from pathlib import Path from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions async def analyze_codebase(directory: str) - dict: 使用 Claude Agent SDK 分析代码库 Args: directory: 要分析的目录路径 Returns: 包含分析结果的字典 # 配置 Agent 选项 options ClaudeAgentOptions( # 只允许读取操作确保安全 allowed_tools[Read, Grep, Glob], # 使用只读模式 permission_modeplan, # 限制执行轮次 max_turns25, # 设置工作目录 cwddirectory, # 使用 Sonnet 模型平衡性能和成本 modelsonnet ) # 构建分析提示 prompt f请分析 {directory} 目录中的代码库。 ## 分析任务 1. **项目结构** - 识别主要目录和文件 - 确定项目类型Web 应用、API、CLI 工具等 - 列出使用的技术栈 2. **代码质量** - 检查代码组织是否合理 - 识别重复代码 - 评估命名规范 3. **潜在问题** - 查找可能的 bug - 识别安全隐患 - 发现性能问题 4. **改进建议** - 提出具体的改进方案 - 优先级排序 ## 输出格式 请以 Markdown 格式输出报告包含上述所有部分。 在每个问题后注明文件和行号。 # 收集结果 result { directory: directory, timestamp: datetime.now().isoformat(), report: [], tools_used: [], metadata: {} } try: async with ClaudeSDKClient(optionsoptions) as client: await client.query(prompt) async for message in client.receive_response(): match message.type: case text: result[report].append(message.text) case tool_use: tool_info f{message.tool_name}: {message.tool_input.get(file_path, message.tool_input.get(pattern, ))} result[tools_used].append(tool_info) print(f [scanning] {tool_info}) case result: result[metadata] { duration_ms: message.duration_ms, total_cost_usd: message.total_cost_usd, num_turns: message.num_turns, input_tokens: message.usage.get(input_tokens, 0), output_tokens: message.usage.get(output_tokens, 0) } case error: print(f [error] {message.error}) result[error] message.error except Exception as e: result[error] str(e) print(fError during analysis: {e}) return result def format_report(result: dict) - str: 格式化分析报告 lines [ * 60, CODE ANALYSIS REPORT, * 60, , fDirectory: {result[directory]}, fTimestamp: {result[timestamp]}, ] if result.get(error): lines.extend([ WARNING: Analysis encountered an error:, result[error], ]) lines.extend([ - * 60, REPORT, - * 60, ]) # 添加报告内容 report_text \n.join(result.get(report, [])) lines.append(report_text) # 添加元数据 if result.get(metadata): meta result[metadata] lines.extend([ , - * 60, STATISTICS, - * 60, fDuration: {meta.get(duration_ms, 0) / 1000:.2f}s, fCost: ${meta.get(total_cost_usd, 0):.4f}, fTurns: {meta.get(num_turns, 0)}, fTokens: {meta.get(input_tokens, 0)} in / {meta.get(output_tokens, 0)} out, * 60 ]) return \n.join(lines) async def main(): 主函数 if len(sys.argv) 2: print(Usage: python code_analyzer.py directory) print(Example: python code_analyzer.py ./src) sys.exit(1) directory sys.argv[1] if not Path(directory).is_dir(): print(fError: {directory} is not a valid directory) sys.exit(1) print(fAnalyzing codebase: {directory}) print( This may take a few minutes...) print() # 运行分析 result await analyze_codebase(directory) # 输出报告 report format_report(result) print(report) # 保存报告到文件 report_file fanalysis-report-{datetime.now().strftime(%Y%m%d-%H%M%S)}.md with open(report_file, w) as f: f.write(report) print(f\nReport saved to: {report_file}) if __name__ __main__: asyncio.run(main())在 Agent 中注入和使用自定义工具from claude_agent_sdk import tool tool( nameget_weather, descriptionGet current weather for a city, parameters{city: str, units: str} ) async def get_weather(args): city args[city] units args.get(units, celsius) # 调用天气 API示例 weather await fetch_weather_api(city, units) return { content: [ {type: text, text: fWeather in {city}: {weather}} ] } from claude_agent_sdk import tool, create_sdk_mcp_server tool(greet, Greet a user by name, {name: str}) async def greet_user(args): return { content: [ {type: text, text: fHello, {args[name]}!} ] } tool(calculate, Perform a calculation, {expression: str}) async def calculate(args): try: result eval(args[expression]) # 生产环境请用安全的表达式解析器 return { content: [ {type: text, text: fResult: {result}} ] } except Exception as e: return { content: [ {type: text, text: fError: {e}} ], isError: True } # 创建 MCP 服务器 server create_sdk_mcp_server( namemy-tools, version1.0.0, tools[greet_user, calculate] )