第 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 自然不会用

本章完整示例代码

以下是本章涉及的完整示例文件，可以直接复制运行。

my_first_tool.py — 计算器工具

import anyio
from typing import Any

from claude_agent_sdk import (
    AssistantMessage,
    ClaudeAgentOptions,
    ClaudeSDKClient,
    ResultMessage,
    TextBlock,
    ToolUseBlock,
    ToolResultBlock,
    create_sdk_mcp_server,
    tool,
)


# ============================================================
# 第 1 步：定义工具
# 每个工具都是一个 async 函数 + @tool 装饰器
# ============================================================

@tool("add", "把两个数相加，返回它们的和", {"a": float, "b": float})
async def add(args: dict[str, Any]) -> dict[str, Any]:
    """加法工具。"""
    result = args["a"] + args["b"]
    return {
        "content": [{"type": "text", "text": f"{args['a']} + {args['b']} = {result}"}]
    }


@tool("subtract", "用第一个数减去第二个数", {"a": float, "b": float})
async def subtract(args: dict[str, Any]) -> dict[str, Any]:
    """减法工具。"""
    result = args["a"] - args["b"]
    return {
        "content": [{"type": "text", "text": f"{args['a']} - {args['b']} = {result}"}]
    }


@tool("multiply", "把两个数相乘，返回它们的积", {"a": float, "b": float})
async def multiply(args: dict[str, Any]) -> dict[str, Any]:
    """乘法工具。"""
    result = args["a"] * args["b"]
    return {
        "content": [{"type": "text", "text": f"{args['a']} * {args['b']} = {result}"}]
    }


@tool("divide", "用第一个数除以第二个数，注意除数不能为零", {"a": float, "b": float})
async def divide(args: dict[str, Any]) -> dict[str, Any]:
    """除法工具，包含除零检查。"""
    # 错误处理：除数为零时返回 is_error
    if args["b"] == 0:
        return {
            "content": [{"type": "text", "text": "错误：除数不能为零！"}],
            "is_error": True,
        }
    result = args["a"] / args["b"]
    return {
        "content": [{"type": "text", "text": f"{args['a']} / {args['b']} = {result}"}]
    }


# ============================================================
# 第 2 步：创建 MCP 服务器
# 把四个工具打包到一个名为 "calculator" 的服务器里
# ============================================================

calculator_server = create_sdk_mcp_server(
    name="calculator",
    version="1.0.0",
    tools=[add, subtract, multiply, divide],
)


# ============================================================
# 第 3 步：使用 ClaudeSDKClient 调用
# ============================================================

async def main():
    """演示 Claude 如何使用计算器工具。"""

    # 配置 Claude，注册计算器服务器
    options = ClaudeAgentOptions(
        # 注册 MCP 服务器，"calc" 是我们给服务器起的短名
        mcp_servers={"calc": calculator_server},
        # 允许 Claude 使用的工具列表
        # 格式：mcp__<服务器短名>__<工具名>
        allowed_tools=[
            "mcp__calc__add",
            "mcp__calc__subtract",
            "mcp__calc__multiply",
            "mcp__calc__divide",
        ],
        # 跳过权限确认，自动执行工具
        permission_mode="bypassPermissions",
        # 最多 5 轮对话（给 Claude 足够的空间使用工具）
        max_turns=5,
    )

    print("=" * 50)
    print("计算器工具示例")
    print("=" * 50)
    print()

    # 让 Claude 做一道需要多步计算的题
    prompt = "请帮我计算：(100 + 50) * 2 / 3，请一步一步算，每步都用工具。"
    print(f"问题：{prompt}\n")

    async with ClaudeSDKClient(options) as client:
        await client.query(prompt)

        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        # Claude 的文字回复
                        print(f"Claude: {block.text}")
                    elif isinstance(block, ToolUseBlock):
                        # Claude 调用了工具
                        print(f"[调用工具] {block.name}({block.input})")
                    elif isinstance(block, ToolResultBlock):
                        # 工具返回的结果
                        print(f"[工具结果] {block.content}")

            elif isinstance(msg, ResultMessage):
                # 对话结束，打印统计信息
                print()
                print("-" * 50)
                print(f"对话轮数: {msg.num_turns}")
                print(f"耗时: {msg.duration_ms}ms")
                if msg.total_cost_usd is not None:
                    print(f"花费: ${msg.total_cost_usd:.4f}")


if __name__ == "__main__":
    anyio.run(main)

运行方式：

python examples/my_first_tool.py

weather_tool.py — 天气查询工具

import anyio
from typing import Any

from claude_agent_sdk import (
    AssistantMessage,
    ClaudeAgentOptions,
    ClaudeSDKClient,
    ResultMessage,
    TextBlock,
    ToolUseBlock,
    create_sdk_mcp_server,
    tool,
)


# ============================================================
# 模拟天气数据
# 实际项目中，这里可以换成：
# - 调用真实天气 API（和风天气、OpenWeatherMap 等）
# - 查数据库
# - 读取本地缓存
# ============================================================

WEATHER_DATA: dict[str, dict[str, Any]] = {
    "北京": {
        "temp": 22,
        "temp_min": 15,
        "temp_max": 28,
        "condition": "晴天",
        "humidity": 45,
        "wind": "北风 3 级",
    },
    "上海": {
        "temp": 26,
        "temp_min": 22,
        "temp_max": 30,
        "condition": "多云",
        "humidity": 68,
        "wind": "东南风 2 级",
    },
    "广州": {
        "temp": 30,
        "temp_min": 26,
        "temp_max": 34,
        "condition": "雷阵雨",
        "humidity": 85,
        "wind": "南风 4 级",
    },
    "深圳": {
        "temp": 29,
        "temp_min": 25,
        "temp_max": 33,
        "condition": "阵雨转多云",
        "humidity": 80,
        "wind": "西南风 3 级",
    },
    "成都": {
        "temp": 24,
        "temp_min": 18,
        "temp_max": 27,
        "condition": "阴天",
        "humidity": 72,
        "wind": "微风",
    },
}


# ============================================================
# 定义天气工具
# ============================================================

@tool("get_weather", "查询指定城市的当前天气信息", {"city": str})
async def get_weather(args: dict[str, Any]) -> dict[str, Any]:
    """查询天气。

    直接访问外部变量 WEATHER_DATA——这就是 SDK MCP 的好处，
    工具函数和应用程序跑在同一个进程里，能直接用模块级别的变量。
    """
    city = args["city"]

    # 查找天气数据
    weather = WEATHER_DATA.get(city)
    if not weather:
        # 城市不在数据里，返回错误
        available = "、".join(WEATHER_DATA.keys())
        return {
            "content": [
                {
                    "type": "text",
                    "text": f"抱歉，没有找到「{city}」的天气数据。"
                    f"目前支持的城市有：{available}",
                }
            ],
            "is_error": True,
        }

    # 格式化天气信息返回给 Claude
    info = (
        f"【{city}天气】\n"
        f"天气状况：{weather['condition']}\n"
        f"当前温度：{weather['temp']}°C\n"
        f"最低温度：{weather['temp_min']}°C\n"
        f"最高温度：{weather['temp_max']}°C\n"
        f"湿度：{weather['humidity']}%\n"
        f"风力：{weather['wind']}"
    )
    return {"content": [{"type": "text", "text": info}]}


@tool(
    "compare_weather",
    "比较两个城市的天气，返回温度和天气状况的对比",
    {"city1": str, "city2": str},
)
async def compare_weather(args: dict[str, Any]) -> dict[str, Any]:
    """比较两个城市的天气。

    展示工具可以有多个参数，并且做更复杂的数据处理。
    """
    city1 = args["city1"]
    city2 = args["city2"]

    weather1 = WEATHER_DATA.get(city1)
    weather2 = WEATHER_DATA.get(city2)

    # 检查两个城市是否都有数据
    missing = []
    if not weather1:
        missing.append(city1)
    if not weather2:
        missing.append(city2)

    if missing:
        return {
            "content": [
                {
                    "type": "text",
                    "text": f"抱歉，没有找到以下城市的天气数据：{'、'.join(missing)}",
                }
            ],
            "is_error": True,
        }

    # 温度差异
    temp_diff = weather1["temp"] - weather2["temp"]
    if temp_diff > 0:
        temp_compare = f"{city1}比{city2}高 {temp_diff}°C"
    elif temp_diff < 0:
        temp_compare = f"{city1}比{city2}低 {abs(temp_diff)}°C"
    else:
        temp_compare = f"两个城市温度相同"

    comparison = (
        f"【{city1} vs {city2} 天气对比】\n"
        f"{city1}：{weather1['condition']}，{weather1['temp']}°C，"
        f"湿度 {weather1['humidity']}%\n"
        f"{city2}：{weather2['condition']}，{weather2['temp']}°C，"
        f"湿度 {weather2['humidity']}%\n"
        f"温度差异：{temp_compare}"
    )
    return {"content": [{"type": "text", "text": comparison}]}


# ============================================================
# 创建天气服务器并运行
# ============================================================

weather_server = create_sdk_mcp_server(
    name="weather",
    version="1.0.0",
    tools=[get_weather, compare_weather],
)


async def main():
    """演示 Claude 使用天气查询工具。"""

    options = ClaudeAgentOptions(
        mcp_servers={"weather": weather_server},
        allowed_tools=[
            "mcp__weather__get_weather",
            "mcp__weather__compare_weather",
        ],
        permission_mode="bypassPermissions",
        max_turns=5,
    )

    print("=" * 50)
    print("天气查询工具示例")
    print("=" * 50)
    print()

    # 问一个需要查天气的问题
    prompt = "今天北京和广州哪个更热？我要出差，帮我看看该带什么衣服。"
    print(f"问题：{prompt}\n")

    async with ClaudeSDKClient(options) as client:
        await client.query(prompt)

        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(block, ToolUseBlock):
                        print(f"  [调用工具] {block.name}({block.input})")

            elif isinstance(msg, ResultMessage):
                print()
                print("-" * 50)
                print(f"对话轮数: {msg.num_turns}")
                if msg.total_cost_usd is not None:
                    print(f"花费: ${msg.total_cost_usd:.4f}")


if __name__ == "__main__":
    anyio.run(main)

运行方式：

python examples/weather_tool.py

multi_tool_server.py — 多工具服务器（待办事项）

这个示例比较长（约 250 行），我们先分步拆解，最后再给完整代码。

第一步：准备"数据库"

在内存里放一个 Python 列表当"数据库"，预置 3 条待办事项，再加一个自增 ID 计数器：

todos: list[dict[str, Any]] = [
    {"id": 1, "title": "买菜", "done": False},
    {"id": 2, "title": "写周报", "done": False},
    {"id": 3, "title": "健身", "done": True},
]

next_id = 4  # 下一条待办的 ID 从 4 开始

就像你在桌上摆了一张纸，上面已经写了 3 件事。

第二步：定义 4 个工具

用 @tool 装饰器注册了 4 个函数。每个函数就是一个"遥控器按钮"，Claude 可以决定什么时候按。

工具名	干什么	注解（给 Claude 的提示）
`list_todos`	查看待办事项，支持按状态筛选	`readOnlyHint=True` — 只读，放心用
`add_todo`	添加一条新待办	`openWorldHint=False` — 只改内存，不联网
`complete_todo`	把某条标记为已完成	`idempotentHint=True` — 重复调也没事
`delete_todo`	永久删除某条	`destructiveHint=True` — 危险操作，谨慎使用

以 list_todos 为例，它用了 JSON Schema 写法（因为 status 参数有枚举限制和默认值）：

@tool(
    "list_todos",
    "列出所有待办事项，可以按状态筛选（all/done/undone）",
    {
        "type": "object",
        "properties": {
            "status": {
                "type": "string",
                "description": "筛选状态：all/done/undone",
                "enum": ["all", "done", "undone"],
                "default": "all",
            },
        },
        "required": [],  # status 是可选的，默认 all
    },
    annotations=ToolAnnotations(readOnlyHint=True),  # 只查不改
)
async def list_todos(args: dict[str, Any]) -> dict[str, Any]:
    status = args.get("status", "all")
    if status == "done":
        filtered = [t for t in todos if t["done"]]
    elif status == "undone":
        filtered = [t for t in todos if not t["done"]]
    else:
        filtered = todos
    # ... 格式化输出并返回

再看 add_todo，它用简单写法（只有一个必填参数），并且直接修改内存中的 todos 列表：

@tool(
    "add_todo",
    "添加一条新的待办事项",
    {"title": str},
    annotations=ToolAnnotations(openWorldHint=False),  # 只改内存，不联网
)
async def add_todo(args: dict[str, Any]) -> dict[str, Any]:
    global next_id
    title = args["title"]
    new_todo = {"id": next_id, "title": title, "done": False}
    todos.append(new_todo)   # 直接操作全局变量！
    next_id += 1
    return {"content": [{"type": "text", "text": f"已添加待办事项 #{new_todo['id']}：{title}"}]}

complete_todo 标记为幂等（多次标记完成效果一样），delete_todo 标记为破坏性操作：

@tool(
    "complete_todo",
    "把一条待办事项标记为已完成",
    {"id": int},
    annotations=ToolAnnotations(idempotentHint=True),  # 幂等：多次调用效果一样
)
async def complete_todo(args: dict[str, Any]) -> dict[str, Any]:
    todo_id = args["id"]
    for t in todos:
        if t["id"] == todo_id:
            t["done"] = True  # 直接修改内存中的数据
            return {
                "content": [
                    {"type": "text", "text": f"已完成待办事项 #{todo_id}：{t['title']}"}
                ]
            }
    # 找不到就返回错误
    return {
        "content": [{"type": "text", "text": f"错误：没有找到 ID 为 {todo_id} 的待办事项"}],
        "is_error": True,
    }


@tool(
    "delete_todo",
    "永久删除一条待办事项，删除后不可恢复",
    {"id": int},
    annotations=ToolAnnotations(destructiveHint=True),  # 破坏性：Claude 会更谨慎
)
async def delete_todo(args: dict[str, Any]) -> dict[str, Any]:
    todo_id = args["id"]
    for i, t in enumerate(todos):
        if t["id"] == todo_id:
            removed = todos.pop(i)  # 从列表中移除，不可恢复
            return {
                "content": [
                    {"type": "text", "text": f"已删除待办事项 #{todo_id}：{removed['title']}"}
                ]
            }
    return {
        "content": [{"type": "text", "text": f"错误：没有找到 ID 为 {todo_id} 的待办事项"}],
        "is_error": True,
    }

关键点：每个工具函数都直接读写内存中的 todos 列表——这就是 SDK MCP 的核心优势，工具和应用状态在同一个进程里。

第三步：打包成服务器

把 4 个工具装进一个叫 "todo" 的 MCP 服务器。就像把 4 个遥控器按钮装进一个遥控器：

todo_server = create_sdk_mcp_server(
    name="todo",
    version="1.0.0",
    tools=[list_todos, add_todo, complete_todo, delete_todo],
)

第四步：配置客户端

options = ClaudeAgentOptions(
    mcp_servers={"todo": todo_server},       # 注册遥控器
    allowed_tools=[                          # 白名单：Claude 可以按哪些按钮
        "mcp__todo__list_todos",
        "mcp__todo__add_todo",
        "mcp__todo__complete_todo",
        "mcp__todo__delete_todo",
    ],
    permission_mode="bypassPermissions",     # 免确认模式
    max_turns=10,                            # 最多 10 轮，防止无限循环
)

第五步：发指令，跑起来

你对 Claude 说一句自然语言指令，它自己拆解成多个步骤，自动决定该调哪些工具、按什么顺序调：

prompt = (
    "帮我管理一下待办事项："
    "1. 先看看现在有哪些未完成的事；"
    "2. 把「写周报」标记为完成；"
    "3. 再添加一条新的「预约牙医」；"
    "4. 最后列出所有事项让我确认。"
)

async with ClaudeSDKClient(options) as client:
    await client.query(prompt)
    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(block, ToolUseBlock):
                    print(f"  [调用工具] {block.name}({block.input})")

最后打印内存中的 todos 列表，验证数据确实被工具修改了——写周报变成已完成，新增了预约牙医。

运行时序列图

下面这张图展示了程序运行时，4 个角色之间的交互流程：

┌──────┐        ┌──────────┐        ┌──────────┐        ┌────────────┐
│ main │        │ SDK客户端 │        │Claude API│        │MCP todo服务│
└──┬───┘        └────┬─────┘        └────┬─────┘        └─────┬──────┘
   │  query(prompt)  │                   │                    │
   │────────────────>│  发送prompt+工具表 │                    │
   │                 │──────────────────>│                    │
   │                 │                   │                    │
   │                 │  要调 list_todos  │                    │
   │                 │<─────────────────│                    │
   │ [调用工具]       │  执行list_todos   │                    │
   │<────────────────│───────────────────────────────────────>│
   │                 │  返回:"买菜,写周报"│                    │
   │                 │<──────────────────────────────────────│
   │                 │  把结果发回Claude  │                    │
   │                 │──────────────────>│                    │
   │                 │                   │                    │
   │                 │ 要调complete_todo │                    │
   │                 │<─────────────────│                    │
   │ [调用工具]       │ 执行complete_todo │                    │
   │<────────────────│───────────────────────────────────────>│
   │                 │ 返回:"已完成#2"   │  写周报.done=True   │
   │                 │<──────────────────────────────────────│
   │                 │  把结果发回Claude  │                    │
   │                 │──────────────────>│                    │
   │                 │                   │                    │
   │                 │  要调 add_todo    │                    │
   │                 │<─────────────────│                    │
   │ [调用工具]       │  执行 add_todo    │                    │
   │<────────────────│───────────────────────────────────────>│
   │                 │  返回:"已添加#4"  │  todos.append(#4)  │
   │                 │<──────────────────────────────────────│
   │                 │  把结果发回Claude  │                    │
   │                 │──────────────────>│                    │
   │                 │                   │                    │
   │                 │  要调 list_todos  │                    │
   │                 │<─────────────────│                    │
   │ [调用工具]       │  执行 list_todos  │                    │
   │<────────────────│───────────────────────────────────────>│
   │                 │  返回:全部4条事项  │                    │
   │                 │<──────────────────────────────────────│
   │                 │  把结果发回Claude  │                    │
   │                 │──────────────────>│                    │
   │                 │                   │                    │
   │                 │  文字回复:汇总结果 │                    │
   │ Claude:汇总     │<─────────────────│                    │
   │<────────────────│                   │                    │
   │                 │                   │                    │
   │ ResultMessage   │                   │                    │
   │<────────────────│                   │                    │
   │                 │                   │                    │
   │ 打印最终内存状态 │                   │                    │
   │ (验证数据改了)   │                   │                    │

核心要点

工具 = 你给 Claude 的能力：你定义函数，Claude 自己决定什么时候调用
MCP 服务器 = 工具的容器：把相关工具打包在一起
Claude 是"大脑"：它理解自然语言指令，自动拆解成多步工具调用
状态在你这边：todos 列表在 Python 内存中，Claude 通过工具间接操作它
Annotations 是"标签"：不影响功能，但影响 Claude 的"态度"——看到 destructiveHint 会更谨慎，看到 readOnlyHint 会更放心

完整代码

完整可运行代码见 examples/multi_tool_server.py

点击展开完整代码

import anyio
from typing import Any

from mcp.types import ToolAnnotations

from claude_agent_sdk import (
    AssistantMessage,
    ClaudeAgentOptions,
    ClaudeSDKClient,
    ResultMessage,
    TextBlock,
    ToolUseBlock,
    create_sdk_mcp_server,
    tool,
)


# ============================================================
# 应用状态：一个简单的待办事项列表
# 工具可以直接读写这个列表——这就是 SDK MCP 的核心优势
# ============================================================

todos: list[dict[str, Any]] = [
    {"id": 1, "title": "买菜", "done": False},
    {"id": 2, "title": "写周报", "done": False},
    {"id": 3, "title": "健身", "done": True},
]

# 自增 ID 计数器
next_id = 4


# ============================================================
# 定义工具，每个都带 ToolAnnotations
# ============================================================

@tool(
    "list_todos",
    "列出所有待办事项，可以按状态筛选（all/done/undone）",
    # 使用 JSON Schema 写法，因为 status 参数有枚举限制和默认值
    {
        "type": "object",
        "properties": {
            "status": {
                "type": "string",
                "description": "筛选状态：all（全部）、done（已完成）、undone（未完成）",
                "enum": ["all", "done", "undone"],
                "default": "all",
            },
        },
        "required": [],  # status 是可选的，默认 all
    },
    # readOnlyHint=True 告诉 Claude：这个工具只查不改，放心用
    annotations=ToolAnnotations(readOnlyHint=True),
)
async def list_todos(args: dict[str, Any]) -> dict[str, Any]:
    """列出待办事项。"""
    status = args.get("status", "all")

    # 根据 status 筛选
    if status == "done":
        filtered = [t for t in todos if t["done"]]
    elif status == "undone":
        filtered = [t for t in todos if not t["done"]]
    else:
        filtered = todos

    if not filtered:
        return {"content": [{"type": "text", "text": "没有找到待办事项。"}]}

    # 格式化输出
    lines = []
    for t in filtered:
        mark = "[x]" if t["done"] else "[ ]"
        lines.append(f"  {mark} #{t['id']} {t['title']}")

    text = f"待办事项（{status}）：\n" + "\n".join(lines)
    return {"content": [{"type": "text", "text": text}]}


@tool(
    "add_todo",
    "添加一条新的待办事项",
    {"title": str},
    # openWorldHint=False: 不与外部世界交互，只改内存数据
    annotations=ToolAnnotations(openWorldHint=False),
)
async def add_todo(args: dict[str, Any]) -> dict[str, Any]:
    """添加待办事项。"""
    global next_id

    title = args["title"]
    new_todo = {"id": next_id, "title": title, "done": False}
    todos.append(new_todo)
    next_id += 1

    return {
        "content": [
            {"type": "text", "text": f"已添加待办事项 #{new_todo['id']}：{title}"}
        ]
    }


@tool(
    "complete_todo",
    "把一条待办事项标记为已完成",
    {"id": int},
    # idempotentHint=True: 多次标记完成，效果一样（幂等）
    annotations=ToolAnnotations(idempotentHint=True),
)
async def complete_todo(args: dict[str, Any]) -> dict[str, Any]:
    """完成待办事项。"""
    todo_id = args["id"]

    for t in todos:
        if t["id"] == todo_id:
            t["done"] = True
            return {
                "content": [
                    {"type": "text", "text": f"已完成待办事项 #{todo_id}：{t['title']}"}
                ]
            }

    return {
        "content": [{"type": "text", "text": f"错误：没有找到 ID 为 {todo_id} 的待办事项"}],
        "is_error": True,
    }


@tool(
    "delete_todo",
    "永久删除一条待办事项，删除后不可恢复",
    {"id": int},
    # destructiveHint=True: 这是一个破坏性操作，Claude 会更谨慎地使用
    annotations=ToolAnnotations(destructiveHint=True),
)
async def delete_todo(args: dict[str, Any]) -> dict[str, Any]:
    """删除待办事项。"""
    todo_id = args["id"]

    for i, t in enumerate(todos):
        if t["id"] == todo_id:
            removed = todos.pop(i)
            return {
                "content": [
                    {
                        "type": "text",
                        "text": f"已删除待办事项 #{todo_id}：{removed['title']}",
                    }
                ]
            }

    return {
        "content": [{"type": "text", "text": f"错误：没有找到 ID 为 {todo_id} 的待办事项"}],
        "is_error": True,
    }


# ============================================================
# 创建服务器：一个服务器里放多个工具
# ============================================================

todo_server = create_sdk_mcp_server(
    name="todo",
    version="1.0.0",
    tools=[list_todos, add_todo, complete_todo, delete_todo],
)


# ============================================================
# 运行示例
# ============================================================

async def main():
    """演示 Claude 使用多工具服务器管理待办事项。"""

    options = ClaudeAgentOptions(
        mcp_servers={"todo": todo_server},
        allowed_tools=[
            "mcp__todo__list_todos",
            "mcp__todo__add_todo",
            "mcp__todo__complete_todo",
            "mcp__todo__delete_todo",
        ],
        permission_mode="bypassPermissions",
        max_turns=10,
    )

    print("=" * 50)
    print("多工具服务器示例：待办事项管理")
    print("=" * 50)
    print()

    # 初始数据
    print("初始待办事项：")
    for t in todos:
        mark = "[x]" if t["done"] else "[ ]"
        print(f"  {mark} #{t['id']} {t['title']}")
    print()

    # 用自然语言让 Claude 管理待办事项
    prompt = (
        "帮我管理一下待办事项："
        "1. 先看看现在有哪些未完成的事；"
        "2. 把「写周报」标记为完成；"
        "3. 再添加一条新的「预约牙医」；"
        "4. 最后列出所有事项让我确认。"
    )
    print(f"指令：{prompt}\n")

    async with ClaudeSDKClient(options) as client:
        await client.query(prompt)

        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(block, ToolUseBlock):
                        print(f"  [调用工具] {block.name}({block.input})")

            elif isinstance(msg, ResultMessage):
                print()
                print("-" * 50)
                print(f"对话轮数: {msg.num_turns}")
                if msg.total_cost_usd is not None:
                    print(f"花费: ${msg.total_cost_usd:.4f}")

    # 验证应用状态确实被修改了
    print()
    print("最终应用状态（Python 内存中）：")
    for t in todos:
        mark = "[x]" if t["done"] else "[ ]"
        print(f"  {mark} #{t['id']} {t['title']}")


if __name__ == "__main__":
    anyio.run(main)

运行方式：

python examples/multi_tool_server.py

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 的使用