第 5 章:自定义工具(SDK MCP)

Claude 很聪明,但它也有不会的事。这一章教你怎么给 Claude "装 App",让它学会新技能。

前提条件

  1. 已完成前几章的学习,熟悉 query()ClaudeSDKClient 的基本用法
  2. 已安装 SDK:uv add claude-agent-sdk

1. 为什么需要自定义工具?

Claude 自带了一堆工具——Read(读文件)、Write(写文件)、Bash(执行命令)、Glob(搜文件)、 Grep(搜内容)等等。这些工具让它能直接操作你的电脑。

但有些事情,Claude 的内置工具是搞不定的:

自定义工具就是给 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}"}]
    }

几个关键点:

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 是个好习惯。


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 不会直接把错误信息扔给用户。它会:

  1. 读懂错误信息 — 理解出了什么问题
  2. 调整策略 — 比如换一种方式调用工具,或者直接告诉用户"这个操作做不了"
  3. 给出解释 — 用自然语言解释为什么失败了,可能还会给出建议

比如上面的除零错误,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,
        }

要点:


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 写法里的 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_weatherquery_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 没有调用我的工具?

检查几个常见原因:

  1. allowed_tools 写对了吗? 格式是 mcp__<服务器短名>__<工具名>,注意是双下划线
  2. description 写清楚了吗? Claude 根据 description 决定是否使用工具
  3. prompt 有暗示吗? 如果你问的问题和工具功能不沾边,Claude 自然不会用

10. 小结

这一章我们学了:

  1. @tool 装饰器 — 三步把 Python 函数变成 Claude 可以调用的工具
  2. create_sdk_mcp_server — 把工具打包成 MCP 服务器
  3. 注册使用 — 通过 mcp_serversallowed_tools 让 Claude 用上你的工具
  4. ToolAnnotations — 给工具加上"性格标签",帮助 Claude 安全决策
  5. 错误处理 — 用 "is_error": True 优雅地告诉 Claude 出问题了
  6. 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 的使用