从现在起，你可以基于 Claude Code SDK 构建自己的智能体了！附大量例子
99% 的 Agent 创业公司，都可以扔掉你的代码了转用 Claude Code SDK 来实现了。
刚刚，Claude Code SDK 发布了支持自定义工具和钩子函数的更新：
Claude Code 开发者 Thariq(@trq212) 宣布：
Claude Code SDK 现在支持直接在代码中使用自定义工具和钩子。此外，我们刷新了所有文档，包含完整的参考和 10 个新指南。
虽然 Claude Code 并不开源，甚至还莫名降智，但这不妨碍它性能强大，我每天都会用来做各种编程之外的事。
而基于 Claude Code SDK，再结合 MCP、工具、命令、钩子等周边功能，可以说，这能帮我们解决绝大多数在电脑上需要做的事情，和绝大多数所谓 Agent 公司们在做的事情。
自定义工具的实现
自定义工具的实现方式很简单：定义一个函数，将其注册为 SDK 的 in-process MCP 服务器，Claude Code 就能调用它。
这种「进程内 MCP」的设计让扩展变得异常优雅。
来看实现代码。
（又是这个获取天气的示例……，好在后面有更多例子）
TypeScript 版本：
import { query, tool, createSdkMcpServer } from"@anthropic-ai/claude-code";
import { z } from"zod";
// 创建一个带有自定义工具的 SDK MCP 服务器
const customServer = createSdkMcpServer({
name: "my-custom-tools",
version: "1.0.0",
tools: [
tool(
"get_weather",
"Get current weather for a location",
{
location: z.string().describe("City name or coordinates"),
units: z.enum(["celsius", "fahrenheit"]).default("celsius").describe("Temperature units")
},
async (args) => {
// 调用天气 API
const response = await fetch(
`https://api.weather.com/v1/current?q=${args.location}&units=${args.units}`
);
const data = await response.json();
return {
content: [{
type: "text",
text: `Temperature: ${data.temp}°\nConditions: ${data.conditions}\nHumidity: ${data.humidity}%`
}]
};
}
)
]
});
Python 版本：
from claude_code_sdk import tool, create_sdk_mcp_server
from typing import Any
import aiohttp
# 使用装饰器定义自定义工具
@tool("get_weather", "Get current weather for a location", {"location": str, "units": str})
asyncdef get_weather(args: dict[str, Any]) -> dict[str, Any]:
units = args.get('units', 'celsius')
asyncwith aiohttp.ClientSession() as session:
asyncwith session.get(
f"https://api.weather.com/v1/current?q={args['location']}&units={units}"
) as response:
data = await response.json()
return {
"content": [{
"type": "text",
"text": f"Temperature: {data['temp']}°\nConditions: {data['conditions']}\nHumidity: {data['humidity']}%"
}]
}
# 创建服务器
custom_server = create_sdk_mcp_server(
name="my-custom-tools",
version="1.0.0",
tools=[get_weather]
)
使用自定义工具
重要提示：自定义 MCP 工具需要流式输入模式。必须使用异步生成器作为 prompt 参数，简单的字符串不能与 MCP 服务器配合使用。
工具名称遵循特定格式：mcp__{server_name}__{tool_name}。例如，my-custom-tools 服务器中的 get_weather 工具会变成 mcp__my-custom-tools__get_weather。
TypeScript 调用示例：
import { query } from"@anthropic-ai/claude-code";
// 使用异步生成器进行流式输入
asyncfunction* generateMessages() {
yield {
type: "user"asconst,
message: {
role: "user"asconst,
content: "What's the weather in San Francisco?"
}
};
}
forawait (const message of query({
prompt: generateMessages(), // 使用异步生成器
options: {
mcpServers: {
"my-custom-tools": customServer // 作为对象传递，不是数组
},
// 可选：指定 Claude 可以使用哪些工具
allowedTools: [
"mcp__my-custom-tools__get_weather",
],
maxTurns: 3
}
})) {
if (message.type === "result" && message.subtype === "success") {
console.log(message.result);
}
}
Python 调用示例：
from claude_code_sdk import query, ClaudeCodeOptions
asyncdef message_generator():
yield {
"type": "user",
"message": {
"role": "user",
"content": "What's the weather in San Francisco?"
}
}
asyncfor message in query(
prompt=message_generator(), # 使用异步生成器
options=ClaudeCodeOptions(
mcp_servers={"my-custom-tools": custom_server},
allowed_tools=[
"mcp__my-custom-tools__get_weather",
],
max_turns=3
)
):
if hasattr(message, 'result'):
print(message.result)
权限系统架构
SDK 提供了四层互补的权限控制机制，让我们能够精确控制工具的使用。
权限模式
default | ||
plan | ||
acceptEdits | ||
bypassPermissions |
设置权限模式的代码：
// 初始配置
const result = await query({
prompt: "Help me refactor this code",
options: {
permissionMode: 'default'
}
});
// 动态更改（仅流式模式）
asyncfunction* streamInput() {
yield {
type: 'user',
message: {
role: 'user',
content: "Let's start with default permissions"
}
};
}
const q = query({
prompt: streamInput(),
options: {
permissionMode: 'default'
}
});
// 动态切换模式
await q.setPermissionMode('acceptEdits');
canUseTool 回调
canUseTool 回调在 Claude Code 需要显示权限提示时触发，可以实现交互式工具批准：
async function promptForToolApproval(toolName: string, input: any) {
console.log("\n🔧 Tool Request:");
console.log(` Tool: ${toolName}`);
// 显示工具参数
if (input && Object.keys(input).length > 0) {
console.log(" Parameters:");
for (const [key, value] of Object.entries(input)) {
let displayValue = value;
if (typeof value === 'string' && value.length > 100) {
displayValue = value.substring(0, 100) + "...";
} elseif (typeof value === 'object') {
displayValue = JSON.stringify(value, null, 2);
}
console.log(` ${key}: ${displayValue}`);
}
}
// 获取用户批准
const approved = await getUserApproval();
if (approved) {
console.log(" ✅ Approved\n");
return {
behavior: "allow",
updatedInput: input
};
} else {
console.log(" ❌ Denied\n");
return {
behavior: "deny",
message: "User denied permission for this tool"
};
}
}
// 使用权限回调
const result = await query({
prompt: "Help me analyze this codebase",
options: {
canUseTool: async (toolName, input) => {
return promptForToolApproval(toolName, input);
}
}
});
钩子函数
钩子提供了对工具执行各个阶段的程序化控制：
const result = await query({
prompt: "Help me refactor this code",
options: {
hooks: {
PreToolUse: [{
hooks: [async (input, toolUseId, { signal }) => {
console.log(`Tool request: ${input.tool_name}`);
// 自定义验证逻辑
if (input.tool_name === "Bash") {
const command = input.tool_input.command;
if (command.startsWith("rm -rf")) {
return {
decision: "block",
reason: "Dangerous command blocked"
};
}
}
return { continue: true };
}]
}],
PostToolUse: [{
hooks: [async (input, toolUseId, { signal }) => {
console.log(`Tool completed: ${input.tool_name}`);
// 记录或审计工具结果
return { continue: true };
}]
}]
}
}
});
Python 版本钩子实现：
from claude_code_sdk import query, ClaudeCodeOptions, HookMatcher, HookContext
asyncdef pre_tool_hook(
input_data: dict[str, Any],
tool_use_id: str | None,
context: HookContext
) -> dict[str, Any]:
print(f"Tool request: {input_data['tool_name']}")
if input_data['tool_name'] == 'Bash':
command = input_data['tool_input'].get('command', '')
if command.startswith('rm -rf'):
return {
'hookSpecificOutput': {
'hookEventName': 'PreToolUse',
'permissionDecision': 'deny',
'permissionDecisionReason': 'Dangerous command blocked'
}
}
return {}
options = ClaudeCodeOptions(
hooks={
'PreToolUse': [
HookMatcher(matcher='Bash', hooks=[pre_tool_hook])
]
}
)
权限规则
在 settings.json 中配置声明式权限规则：
{
"permissions": {
"allow": [
"Bash(npm run lint)",
"Bash(npm run test:*)",
"Read(~/.zshrc)"
],
"deny": [
"Bash(curl:*)",
"Read(./.env)",
"Read(./secrets/**)",
"WebFetch"
],
"ask": [
"Bash(git push:*)",
"Write(./production/**)"
]
}
}
规则语法说明：
Bash 规则：使用前缀匹配。例如
Bash(npm:*)匹配任何以 "npm" 开头的命令文件规则：支持 glob 模式。例如
Read(./src/**/*.ts)匹配 src 目录下的所有 TypeScript 文件工具规则：省略括号控制整个工具。例如
WebFetch阻止所有网络获取
实用工具示例
数据库查询工具
const databaseServer = createSdkMcpServer({
name: "database-tools",
version: "1.0.0",
tools: [
tool(
"query_database",
"Execute a database query",
{
query: z.string().describe("SQL query to execute"),
params: z.array(z.any()).optional().describe("Query parameters")
},
async (args) => {
const results = await db.query(args.query, args.params || []);
return {
content: [{
type: "text",
text: `Found ${results.length} rows:\n${JSON.stringify(results, null, 2)}`
}]
};
}
)
]
});
API 网关工具
这个工具可以调用多个外部服务：
const apiGatewayServer = createSdkMcpServer({
name: "api-gateway",
version: "1.0.0",
tools: [
tool(
"api_request",
"Make authenticated API requests to external services",
{
service: z.enum(["stripe", "github", "openai", "slack"]).describe("Service to call"),
endpoint: z.string().describe("API endpoint path"),
method: z.enum(["GET", "POST", "PUT", "DELETE"]).describe("HTTP method"),
body: z.record(z.any()).optional().describe("Request body"),
query: z.record(z.string()).optional().describe("Query parameters")
},
async (args) => {
const config = {
stripe: { baseUrl: "https://api.stripe.com/v1", key: process.env.STRIPE_KEY },
github: { baseUrl: "https://api.github.com", key: process.env.GITHUB_TOKEN },
openai: { baseUrl: "https://api.openai.com/v1", key: process.env.OPENAI_KEY },
slack: { baseUrl: "https://slack.com/api", key: process.env.SLACK_TOKEN }
};
const { baseUrl, key } = config[args.service];
const url = new URL(`${baseUrl}${args.endpoint}`);
if (args.query) {
Object.entries(args.query).forEach(([k, v]) => url.searchParams.set(k, v));
}
const response = await fetch(url, {
method: args.method,
headers: {
Authorization: `Bearer ${key}`,
"Content-Type": "application/json"
},
body: args.body ? JSON.stringify(args.body) : undefined
});
const data = await response.json();
return {
content: [{
type: "text",
text: JSON.stringify(data, null, 2)
}]
};
}
)
]
});
计算器工具
包含基础运算和复利计算：
const calculatorServer = createSdkMcpServer({
name: "calculator",
version: "1.0.0",
tools: [
tool(
"calculate",
"Perform mathematical calculations",
{
expression: z.string().describe("Mathematical expression to evaluate"),
precision: z.number().optional().default(2).describe("Decimal precision")
},
async (args) => {
try {
// 生产环境使用安全的数学计算库
const result = eval(args.expression); // 仅示例！
const formatted = Number(result).toFixed(args.precision);
return {
content: [{
type: "text",
text: `${args.expression} = ${formatted}`
}]
};
} catch (error) {
return {
content: [{
type: "text",
text: `Error: Invalid expression - ${error.message}`
}]
};
}
}
),
tool(
"compound_interest",
"Calculate compound interest for an investment",
{
principal: z.number().positive().describe("Initial investment amount"),
rate: z.number().describe("Annual interest rate (as decimal)"),
time: z.number().positive().describe("Investment period in years"),
n: z.number().positive().default(12).describe("Compounding frequency per year")
},
async (args) => {
const amount = args.principal * Math.pow(1 + args.rate / args.n, args.n * args.time);
const interest = amount - args.principal;
return {
content: [{
type: "text",
text: `Investment Analysis:\n` +
`Principal: $${args.principal.toFixed(2)}\n` +
`Rate: ${(args.rate * 100).toFixed(2)}%\n` +
`Time: ${args.time} years\n` +
`Compounding: ${args.n} times per year\n\n` +
`Final Amount: $${amount.toFixed(2)}\n` +
`Interest Earned: $${interest.toFixed(2)}\n` +
`Return: ${((interest / args.principal) * 100).toFixed(2)}%`
}]
};
}
)
]
});
多工具协同
当 MCP 服务器有多个工具时，可以选择性地允许它们：
const multiToolServer = createSdkMcpServer({
name: "utilities",
version: "1.0.0",
tools: [
tool("calculate", "Perform calculations", { /* ... */ }, async (args) => { /* ... */ }),
tool("translate", "Translate text", { /* ... */ }, async (args) => { /* ... */ }),
tool("search_web", "Search the web", { /* ... */ }, async (args) => { /* ... */ })
]
});
// 只允许特定工具
asyncfunction* generateMessages() {
yield {
type: "user"asconst,
message: {
role: "user"asconst,
content: "Calculate 5 + 3 and translate 'hello' to Spanish"
}
};
}
forawait (const message of query({
prompt: generateMessages(),
options: {
mcpServers: {
utilities: multiToolServer
},
allowedTools: [
"mcp__utilities__calculate", // 允许计算器
"mcp__utilities__translate", // 允许翻译器
// "mcp__utilities__search_web" 不被允许
]
}
})) {
// 处理消息
}
错误处理
优雅地处理错误，可以提供有意义的反馈：
tool(
"fetch_data",
"Fetch data from an API",
{
endpoint: z.string().url().describe("API endpoint URL")
},
async (args) => {
try {
const response = await fetch(args.endpoint);
if (!response.ok) {
return {
content: [{
type: "text",
text: `API error: ${response.status} ${response.statusText}`
}]
};
}
const data = await response.json();
return {
content: [{
type: "text",
text: JSON.stringify(data, null, 2)
}]
};
} catch (error) {
return {
content: [{
type: "text",
text: `Failed to fetch data: ${error.message}`
}]
};
}
}
)
Python 类型安全
Python 版本支持多种模式定义方法：
# 简单类型映射 - 推荐用于大多数情况
@tool(
"process_data",
"Process structured data with type safety",
{
"name": str,
"age": int,
"email": str,
"preferences": list # 可选参数可以在函数中处理
}
)
asyncdef process_data(args: dict[str, Any]) -> dict[str, Any]:
name = args["name"]
age = args["age"]
email = args["email"]
preferences = args.get("preferences", [])
print(f"Processing {name}'s data (age: {age})")
return {
"content": [{
"type": "text",
"text": f"Processed data for {name}"
}]
}
# 对于更复杂的模式，可以使用 JSON Schema 格式
@tool(
"advanced_process",
"Process data with advanced validation",
{
"type": "object",
"properties": {
"name": {"type": "string"},
"age": {"type": "integer", "minimum": 0, "maximum": 150},
"email": {"type": "string", "format": "email"},
"format": {"type": "string", "enum": ["json", "csv", "xml"], "default": "json"}
},
"required": ["name", "age", "email"]
}
)
asyncdef advanced_process(args: dict[str, Any]) -> dict[str, Any]:
return {
"content": [{
"type": "text",
"text": f"Advanced processing for {args['name']}"
}]
}
完整功能支持
SDK 提供了对所有 Claude Code 默认功能的访问，通过相同的文件系统配置：
子智能体：存储在
./.claude/agents/的专用智能体钩子：在
./.claude/settings.json中配置的自定义命令斜杠命令：定义在
./.claude/commands/的 Markdown 文件记忆（CLAUDE.md）：通过
CLAUDE.md文件维护项目上下文
这些功能通过读取相同的文件系统位置来工作，与 Claude Code 对应功能完全一致。
可以构建什么
根据文档，基于 Claude Code SDK 我们可以构建各类 Agent，包括但不限于：
编码智能体：
SRE 智能体诊断和修复生产问题
安全审查机器人审计代码漏洞
值班工程助手分类事件
代码审查智能体执行风格和最佳实践
业务智能体：
法律助手审查合同和合规性
财务顾问分析报告和预测
客户支持智能体解决技术问题
营销团队的内容创建助手
现在，所有这些 Agent 都可以通过 Claude Code SDK 轻松实现，而不需要从零开始构建复杂的基础设施了。
封装一下业务 API，弄弄数据，搞搞 UI，就可以了。
顺便，我还让 Claude 给做了个 《基于 Claude Code SDK 开发 Agent》的PPT，贴上来供参考：
自定义工具文档: https://docs.anthropic.com/en/docs/claude-code/sdk/custom-tools
[2]SDK 权限文档: https://docs.anthropic.com/en/docs/claude-code/sdk/sdk-permissions
[3]SDK 概览: https://docs.anthropic.com/en/docs/claude-code/sdk/sdk-overview
[4]子智能体指南: https://docs.anthropic.com/en/docs/claude-code/sdk/subagents
[5]斜杠命令指南: https://docs.anthropic.com/en/docs/claude-code/sdk/sdk-slash-commands
[6]TypeScript SDK 参考: https://docs.anthropic.com/en/docs/claude-code/sdk/typescript
[7]Python SDK 参考: https://docs.anthropic.com/en/docs/claude-code/sdk/python
[8]MCP 文档: https://modelcontextprotocol.io
[9]SDK 配置: https://docs.anthropic.com/en/docs/claude-code/sdk/sdk-configuration
👇
👇
👇
另外，我还用AI 进行了全网的AI 资讯采集，并用AI 进行挑选、审核、翻译、总结后发布到《AGI Hunt》的实时AI 快讯群中。
这是个只有信息、没有感情的 AI 资讯信息流（不是推荐流、不卖课、不讲道理、不教你做人、只提供信息、希望能为你节省一些时间）
欢迎加入！
也欢迎加群和5000+群友交流。
