第 5 章:自定义工具(SDK MCP)
Claude 很聪明,但它也有不会的事。这一章教你怎么给 Claude "装 App",让它学会新技能。
前提条件
- 已完成前几章的学习,熟悉
query()和ClaudeSDKClient的基本用法- 已安装 SDK:
uv add claude-agent-sdk
1. 为什么需要自定义工具?
Claude 自带了一堆工具——Read(读文件)、Write(写文件)、Bash(执行命令)、Glob(搜文件)、 Grep(搜内容)等等。这些工具让它能直接操作你的电脑。
但有些事情,Claude 的内置工具是搞不定的:
- 查天气?它不会。
- 查你的数据库?它不会。
- 调用你们公司内部的 API?它不会。
- 操作你自己的业务系统?它还是不会。
自定义工具就是给 Claude "装 App"。
就像你的手机出厂时自带了计算器、日历、相机,但你还是要去应用商店装微信、淘宝、 高德地图——自定义工具就是让 Claude 也能用上"你开发的 App"。
装了天气工具,Claude 就能帮你查天气。 装了数据库工具,Claude 就能帮你查数据。 装了什么工具,Claude 就会什么技能。
2. SDK MCP vs 传统 MCP
在深入代码之前,先说一个概念:MCP(Model Context Protocol)。
MCP 是 Anthropic 定义的一套标准协议,用来给 AI 模型"安装"工具。 你不用深究协议细节,只要知道:通过 MCP,你可以把自己写的函数变成 Claude 可以调用的工具。
MCP 有两种用法:
传统 MCP:像叫外卖
传统方式下,你的工具跑在一个独立的进程里。Claude 想用这个工具时,得通过 stdin/stdout 跟它通信。流程是这样的:
Claude: "我想查个天气" → 发消息给工具进程 → 工具进程处理 → 返回结果给 Claude
就像叫外卖:你下单,等骑手取餐、送餐,最后才能吃上。能吃,但中间多了很多环节。
SDK MCP:像自己下厨
SDK MCP 不一样。你的工具直接跑在你的 Python 进程里,和你的主程序在同一个进程中。 Claude 调用工具时,就是直接调用你写的 Python 函数,没有进程间通信的开销。
Claude: "我想查个天气" → 直接调用你的 Python 函数 → 立刻拿到结果
就像自己下厨:打开冰箱,拿出食材,马上就能做。
SDK MCP 的优势
| 对比项 | 传统 MCP | SDK MCP |
|---|---|---|
| 运行方式 | 独立进程 | 同进程 |
| 通信开销 | 有(stdin/stdout) | 无 |
| 部署复杂度 | 需要管理额外进程 | 就是普通 Python 代码 |
| 访问应用状态 | 需要 IPC | 直接访问变量 |
| 调试难度 | 需要跨进程调试 | 正常断点调试 |
本章只讲 SDK MCP。 因为它更简单、更快、更适合绝大多数场景。
3. 三步定义一个工具
好了,概念讲完了,直接上代码。
定义一个自定义工具只需要三步:
Step 1:写一个 async 函数
先写一个普通的异步函数,它接收一个字典参数,返回一个字典结果:
async def add_numbers(args: dict) -> dict:
"""把两个数加起来。"""
result = args["a"] + args["b"]
return {
"content": [{"type": "text", "text": f"计算结果:{result}"}]
}
几个关键点:
- 必须是
async def:因为 MCP 协议要求异步处理 - 参数是一个字典:Claude 调用工具时,会把参数打包成
{"a": 1, "b": 2}这样的字典传进来 - 返回值有固定格式:必须包含
"content"字段,里面是一个列表,每个元素是{"type": "text", "text": "..."}的格式
Step 2:加上 @tool 装饰器
光有函数还不够,Claude 不知道这个函数叫什么、干什么、需要什么参数。
用 @tool 装饰器给它加上"说明书":
from claude_agent_sdk import tool
@tool("add", "把两个数相加", {"a": float, "b": float})
async def add_numbers(args: dict) -> dict:
result = args["a"] + args["b"]
return {
"content": [{"type": "text", "text": f"计算结果:{result}"}]
}
@tool 装饰器接收三个参数:
| 参数 | 说明 | 示例 |
|---|---|---|
name |
工具名称,Claude 用这个名字来调用它 | "add" |
description |
工具描述,帮助 Claude 理解什么时候该用这个工具 | "把两个数相加" |
input_schema |
参数类型定义,告诉 Claude 这个工具需要什么参数 | {"a": float, "b": float} |
description 很重要! Claude 是根据 description 来决定什么时候使用你的工具的。 写得好,Claude 就能准确地在该用的时候用;写得烂,Claude 可能该用的时候不用,不该用的时候乱用。
Step 3:返回标准格式
工具的返回值必须是这个格式:
# 成功时
{
"content": [
{"type": "text", "text": "这里是返回给 Claude 的结果"}
]
}
# 出错时(后面会详细讲)
{
"content": [
{"type": "text", "text": "出错了:除数不能为零"}
],
"is_error": True
}
content 是一个列表,里面可以有多个内容块。最常用的就是 text 类型。
Claude 收到这个结果后,会根据内容来继续对话。
三步就完事了。 现在你有了一个完整的自定义工具。
4. 打包成服务器
一个工具定义好了,但还不能直接给 Claude 用。 你需要把它"打包"成一个 MCP 服务器,然后注册到 Claude 的配置里。
创建服务器
from claude_agent_sdk import create_sdk_mcp_server
# 创建一个名为 "calculator" 的 MCP 服务器,把 add_numbers 工具放进去
calculator_server = create_sdk_mcp_server(
name="calculator",
version="1.0.0",
tools=[add_numbers] # 可以放多个工具
)
create_sdk_mcp_server 就三个参数:
| 参数 | 说明 | 默认值 |
|---|---|---|
name |
服务器名称 | 必填 |
version |
版本号 | "1.0.0" |
tools |
工具列表 | None |
注册到 ClaudeAgentOptions
服务器创建好了,接下来告诉 Claude 去用它:
from claude_agent_sdk import ClaudeAgentOptions
options = ClaudeAgentOptions(
# 注册 MCP 服务器,key 是你给这个服务器起的短名
mcp_servers={"calc": calculator_server},
# 允许 Claude 使用哪些工具
allowed_tools=["mcp__calc__add"],
)
allowed_tools 的命名规则
注意 allowed_tools 里工具名的格式:mcp__<服务器短名>__<工具名>
这里用的是双下划线 __ 分隔,不是单下划线。拆开看就是:
mcp__calc__add
│ │ │
│ │ └── 工具名(@tool 装饰器里的 name)
│ └── 服务器短名(mcp_servers 字典的 key)
└── 固定前缀,表示这是一个 MCP 工具
所以如果你的服务器短名是 "weather",工具名是 "get_forecast",
那完整名称就是 mcp__weather__get_forecast。
完整流程
把上面所有步骤串起来:
import anyio
from claude_agent_sdk import (
AssistantMessage,
ClaudeAgentOptions,
ClaudeSDKClient,
TextBlock,
ResultMessage,
create_sdk_mcp_server,
tool,
)
# ---- 第 1 步:定义工具 ----
@tool("add", "把两个数相加", {"a": float, "b": float})
async def add_numbers(args: dict) -> dict:
result = args["a"] + args["b"]
return {
"content": [{"type": "text", "text": f"计算结果:{result}"}]
}
# ---- 第 2 步:创建服务器 ----
calculator_server = create_sdk_mcp_server(
name="calculator",
tools=[add_numbers]
)
# ---- 第 3 步:配置并调用 ----
async def main():
options = ClaudeAgentOptions(
mcp_servers={"calc": calculator_server},
allowed_tools=["mcp__calc__add"],
permission_mode="bypassPermissions",
max_turns=3,
)
async with ClaudeSDKClient(options) as client:
await client.query("请帮我算一下 123.45 + 678.9 等于多少")
async for msg in client.receive_response():
if isinstance(msg, AssistantMessage):
for block in msg.content:
if isinstance(block, TextBlock):
print(f"Claude: {block.text}")
elif isinstance(msg, ResultMessage):
print(f"\n花费: ${msg.total_cost_usd}")
anyio.run(main)
运行后,Claude 会自动识别出这是一个加法问题,调用你定义的 add 工具,
然后把结果告诉你。
5. 工具注解 ToolAnnotations
有时候你想给 Claude 一些额外的提示,告诉它这个工具的"性格"。 比如:这个工具只读不写?这个工具会删东西?
这就是 ToolAnnotations 的作用。
from claude_agent_sdk import tool
from mcp.types import ToolAnnotations
@tool(
"get_user_count",
"查询系统中的用户总数",
{"table": str},
annotations=ToolAnnotations(
readOnlyHint=True, # 告诉 Claude:这个工具只读,不会改任何东西
),
)
async def get_user_count(args: dict) -> dict:
# 模拟数据库查询
return {
"content": [{"type": "text", "text": "当前用户数:42"}]
}
常用注解
| 注解 | 类型 | 含义 |
|---|---|---|
readOnlyHint |
bool |
只读工具,不会修改任何数据 |
destructiveHint |
bool |
破坏性工具,可能删除或修改重要数据 |
idempotentHint |
bool |
幂等工具,多次调用结果一样(比如"查询余额"调 10 次结果都一样) |
openWorldHint |
bool |
会与外部世界交互(比如发邮件、调第三方 API) |
什么时候用注解?
注解不是必须的,加了也不会改变工具的执行逻辑。它的作用是帮助 Claude 更安全地决策。
举个例子:
- 一个标记了
readOnlyHint=True的工具,Claude 会更放心地使用,因为它知道不会有副作用 - 一个标记了
destructiveHint=True的工具,Claude 会更谨慎,可能会在调用前多确认一下
建议: 如果你的工具涉及写入、删除等操作,最好加上注解。
如果只是查询类的工具,readOnlyHint=True 是个好习惯。
6. 错误处理
工具不可能永远运行成功。网络断了、参数不对、数据库挂了……各种意外都可能发生。
这时候,你需要在返回结果里加上 "is_error": True,告诉 Claude "工具出错了":
@tool("divide", "除法运算", {"a": float, "b": float})
async def divide(args: dict) -> dict:
if args["b"] == 0:
# 告诉 Claude 出错了,并说明原因
return {
"content": [{"type": "text", "text": "错误:除数不能为零!"}],
"is_error": True,
}
result = args["a"] / args["b"]
return {
"content": [{"type": "text", "text": f"计算结果:{result}"}]
}
Claude 收到错误后会怎么做?
Claude 不会直接把错误信息扔给用户。它会:
- 读懂错误信息 — 理解出了什么问题
- 调整策略 — 比如换一种方式调用工具,或者直接告诉用户"这个操作做不了"
- 给出解释 — 用自然语言解释为什么失败了,可能还会给出建议
比如上面的除零错误,Claude 可能会回复:
"抱歉,除数不能为零。你确认一下第二个数字是不是写错了?"
错误处理的最佳实践
@tool("query_db", "查询数据库", {"sql": str})
async def query_db(args: dict) -> dict:
try:
# 实际的数据库查询逻辑
result = execute_query(args["sql"])
return {
"content": [{"type": "text", "text": f"查询结果:{result}"}]
}
except ConnectionError:
return {
"content": [{"type": "text", "text": "数据库连接失败,请稍后重试"}],
"is_error": True,
}
except Exception as e:
return {
"content": [{"type": "text", "text": f"查询出错:{e}"}],
"is_error": True,
}
要点:
- 用 try/except 包裹可能出错的代码,不要让异常直接抛出
- 错误信息要对 Claude 有用,写清楚出了什么问题,Claude 才能做出合理的反应
- 不要直接暴露堆栈信息给 Claude,简洁明了的一句话就够了
7. input_schema 的两种写法
前面一直用的是简单写法 {"a": float, "b": float}。
实际上 input_schema 支持两种格式。
简单写法:类型映射字典
@tool("greet", "和用户打招呼", {"name": str, "age": int})
async def greet(args: dict) -> dict:
return {
"content": [{"type": "text", "text": f"你好,{args['name']}!你 {args['age']} 岁了。"}]
}
这种写法最直观。支持的类型有:
| Python 类型 | 转换后的 JSON Schema 类型 |
|---|---|
str |
"string" |
int |
"integer" |
float |
"number" |
bool |
"boolean" |
SDK 会自动把它转换成标准的 JSON Schema。比如 {"name": str, "age": int} 会变成:
{
"type": "object",
"properties": {
"name": {"type": "string"},
"age": {"type": "integer"}
},
"required": ["name", "age"]
}
注意: 简单写法下,所有参数都是必填的(required)。
JSON Schema 写法:完全自定义
如果你需要更精细的控制——比如参数可选、有默认值、有枚举限制——就直接写 JSON Schema:
@tool(
"search",
"搜索文章",
{
"type": "object",
"properties": {
"keyword": {
"type": "string",
"description": "搜索关键词",
},
"limit": {
"type": "integer",
"description": "返回结果数量上限",
"default": 10,
},
"category": {
"type": "string",
"description": "文章分类",
"enum": ["tech", "science", "art"],
},
},
"required": ["keyword"], # 只有 keyword 是必填的
},
)
async def search(args: dict) -> dict:
keyword = args["keyword"]
limit = args.get("limit", 10) # 有默认值
category = args.get("category") # 可选参数
# ... 搜索逻辑
return {
"content": [{"type": "text", "text": f"搜索 '{keyword}' 找到 {limit} 条结果"}]
}
怎么选?
- 参数少、都是必填的 → 用简单写法,省事
- 参数有可选、有枚举、有描述 → 用 JSON Schema 写法,更灵活
JSON Schema 写法里的 description 字段也很有用——它能帮助 Claude 理解每个参数的含义,
让工具调用更准确。
8. 实战案例
光看理论没用,来几个真实场景。
案例 1:计算器
最经典的入门示例。四则运算,一个服务器搞定:
@tool("add", "加法", {"a": float, "b": float})
async def add(args: dict) -> dict:
return {"content": [{"type": "text", "text": f"{args['a'] + args['b']}"}]}
@tool("subtract", "减法", {"a": float, "b": float})
async def subtract(args: dict) -> dict:
return {"content": [{"type": "text", "text": f"{args['a'] - args['b']}"}]}
@tool("multiply", "乘法", {"a": float, "b": float})
async def multiply(args: dict) -> dict:
return {"content": [{"type": "text", "text": f"{args['a'] * args['b']}"}]}
@tool("divide", "除法", {"a": float, "b": float})
async def divide(args: dict) -> dict:
if args["b"] == 0:
return {"content": [{"type": "text", "text": "错误:除数不能为零"}], "is_error": True}
return {"content": [{"type": "text", "text": f"{args['a'] / args['b']}"}]}
server = create_sdk_mcp_server(name="calculator", tools=[add, subtract, multiply, divide])
完整可运行代码见
examples/my_first_tool.py
案例 2:天气查询
模拟一个天气查询工具。不调真实 API,用字典模拟数据,重点看"工具访问外部数据"的模式:
# 模拟天气数据(实际项目中可以换成真正的 API 调用)
WEATHER_DATA = {
"北京": {"temp": 22, "condition": "晴天", "humidity": 45},
"上海": {"temp": 26, "condition": "多云", "humidity": 68},
"广州": {"temp": 30, "condition": "雷阵雨", "humidity": 85},
}
@tool("get_weather", "查询指定城市的天气信息", {"city": str})
async def get_weather(args: dict) -> dict:
city = args["city"]
weather = WEATHER_DATA.get(city)
if not weather:
return {
"content": [{"type": "text", "text": f"抱歉,没有找到 {city} 的天气数据"}],
"is_error": True,
}
return {
"content": [{"type": "text", "text": (
f"{city}天气:{weather['condition']},"
f"温度 {weather['temp']}°C,"
f"湿度 {weather['humidity']}%"
)}]
}
完整可运行代码见
examples/weather_tool.py
案例 3:多工具服务器 + 注解
一个包含多种工具的服务器,展示 ToolAnnotations 的实际用法:
from mcp.types import ToolAnnotations
@tool(
"list_todos",
"列出所有待办事项",
{"status": str},
annotations=ToolAnnotations(readOnlyHint=True),
)
async def list_todos(args: dict) -> dict:
# 只读操作,标记了 readOnlyHint
...
@tool(
"delete_todo",
"删除一条待办事项",
{"id": int},
annotations=ToolAnnotations(destructiveHint=True),
)
async def delete_todo(args: dict) -> dict:
# 破坏性操作,标记了 destructiveHint
...
完整可运行代码见
examples/multi_tool_server.py
9. 常见问题
Q: 工具名能用中文吗?
不建议。工具名最好用英文和下划线,比如 get_weather、query_db。
描述(description)可以用中文。
Q: 一个服务器能放多少个工具?
没有硬性限制,但建议不要太多。把相关的工具放在同一个服务器里,不相关的分开。 比如"计算器服务器"放加减乘除,"天气服务器"放天气查询。
Q: 工具函数里能访问外部变量吗?
能!这就是 SDK MCP 的一大优势——你的工具跑在同一个 Python 进程里, 可以直接访问应用程序的变量、数据库连接、配置对象等等。
# 全局的数据库连接
db = Database("sqlite:///my.db")
@tool("query", "执行数据库查询", {"sql": str})
async def query_tool(args: dict) -> dict:
# 直接用外部变量 db,不需要任何特殊处理
result = await db.execute(args["sql"])
return {"content": [{"type": "text", "text": str(result)}]}
Q: 工具可以返回图片吗?
可以。content 列表里除了 text 类型,还支持 image 类型:
return {
"content": [
{
"type": "image",
"data": base64_encoded_image,
"mimeType": "image/png",
}
]
}
Q: 为什么 Claude 没有调用我的工具?
检查几个常见原因:
allowed_tools写对了吗? 格式是mcp__<服务器短名>__<工具名>,注意是双下划线- description 写清楚了吗? Claude 根据 description 决定是否使用工具
- prompt 有暗示吗? 如果你问的问题和工具功能不沾边,Claude 自然不会用
10. 小结
这一章我们学了:
- @tool 装饰器 — 三步把 Python 函数变成 Claude 可以调用的工具
- create_sdk_mcp_server — 把工具打包成 MCP 服务器
- 注册使用 — 通过
mcp_servers和allowed_tools让 Claude 用上你的工具 - ToolAnnotations — 给工具加上"性格标签",帮助 Claude 安全决策
- 错误处理 — 用
"is_error": True优雅地告诉 Claude 出问题了 - input_schema 两种写法 — 简单类型映射 vs 完整 JSON Schema
自定义工具是 Claude Agent SDK 最强大的能力之一。掌握了它, 你就能让 Claude 做任何你的 Python 代码能做的事情。
下一章,我们会学习 Hooks(钩子)——在 Claude 执行工具前后插入你自己的逻辑, 实现更精细的控制。
本章文件清单
05-自定义工具/
README.md # 你正在读的这个文件
examples/
my_first_tool.py # 计算器工具:从定义到使用的完整示例
weather_tool.py # 天气查询工具:工具访问外部数据的模式
multi_tool_server.py # 多工具服务器 + ToolAnnotations 的使用