Claude Code SDK #9：自定义工具全解——@tool 装饰器 × in-process MCP × 错误处理 × 非文本返回，把任意函数变成 Claude 的能力

一个工具由四部分组成

Python 实战：天气工具

from typing import Any
import httpx
from claude_agent_sdk import tool, create_sdk_mcp_server

@tool(
    "get_temperature",
    "Get the current temperature at a location",
    {"latitude": float, "longitude": float},
)
async def get_temperature(args: dict[str, Any]) -> dict[str, Any]:
    async with httpx.AsyncClient() as client:
        response = await client.get(
            "https://api.open-meteo.com/v1/forecast",
            params={
                "latitude": args["latitude"],
                "longitude": args["longitude"],
                "current": "temperature_2m",
                "temperature_unit": "fahrenheit",
            },
        )
        data = response.json()
        return {
            "content": [
                {
                    "type": "text",
                    "text": f"Temperature: {data['current']['temperature_2m']}°F",
                }
            ]
        }

# 打包成 in-process MCP server
weather_server = create_sdk_mcp_server(
    name="weather",
    version="1.0.0",
    tools=[get_temperature],
)

注册工具 + allowedTools 命名规范

mcp__{server_name}__{tool_name}

from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage

async def main():
    options = ClaudeAgentOptions(
        mcp_servers={"weather": weather_server},
        allowed_tools=["mcp__weather__get_temperature"],
    )
    async for message in query(
        prompt="What's the temperature in San Francisco?",
        options=options,
    ):
        if isinstance(message, ResultMessage) and message.subtype == "success":
            print(message.result)

readOnlyHint：让 Claude 并行调用工具

from claude_agent_sdk import tool, ToolAnnotations

@tool(
    "get_temperature",
    "Get the current temperature at a location",
    {"latitude": float, "longitude": float},
    annotations=ToolAnnotations(readOnlyHint=True),  # 允许并行
)
async def get_temperature(args):
    return {"content": [{"type": "text", "text": "..."}]}

错误处理：throw vs is_error

@tool("fetch_data", "Fetch data from an API", {"endpoint": str})
async def fetch_data(args: dict[str, Any]) -> dict[str, Any]:
    try:
        async with httpx.AsyncClient() as client:
            response = await client.get(args["endpoint"])
            if response.status_code != 200:
                return {
                    "content": [{
                        "type": "text",
                        "text": f"API error: {response.status_code} {response.reason_phrase}",
                    }],
                    "is_error": True,  # 通知 Claude 这次调用失败了
                }
            data = response.json()
            return {"content": [{"type": "text", "text": json.dumps(data, indent=2)}]}
    except Exception as e:
        return {
            "content": [{"type": "text", "text": f"Failed to fetch data: {str(e)}"}],
            "is_error": True,
        }

返回图片和结构化数据

import base64
import httpx

@tool("fetch_image", "Fetch an image from a URL", {"url": str})
async def fetch_image(args):
    async with httpx.AsyncClient() as client:
        response = await client.get(args["url"])
        return {
            "content": [{
                "type": "image",
                "data": base64.b64encode(response.content).decode("ascii"),
                "mimeType": response.headers.get("content-type", "image/png"),
            }]
        }

{
    "type": "resource",
    "resource": {
        "uri": "file:///tmp/report.md",  # Claude 引用这个标签，不是 SDK 实际路径
        "mimeType": "text/markdown",
        "text": "# Report\n..."           # 实际内容内联
    }
}

处理枚举参数：JSON Schema 全写法

@tool(
    "convert_units",
    "Convert a value from one unit to another",
    {
        "type": "object",
        "properties": {
            "unit_type": {
                "type": "string",
                "enum": ["length", "temperature", "weight"],  # 枚举约束
                "description": "Category of unit",
            },
            "from_unit": {"type": "string", "description": "Unit to convert from"},
            "to_unit": {"type": "string", "description": "Unit to convert to"},
            "value": {"type": "number", "description": "Value to convert"},
        },
        "required": ["unit_type", "from_unit", "to_unit", "value"],
    },
)
async def convert_units(args: dict[str, Any]) -> dict[str, Any]:
    ...

五条实践建议

1. Description 写具体用途，不写工具名。"Get weather data" 不如 "Get the current temperature at a latitude/longitude coordinate in Fahrenheit"，Claude 在多工具场景下选择更准确。
2. 只读工具加 readOnlyHint=True。一个常见场景是同时查多个地点温度——标记后 Claude 会批量并行发起，不需要串行等待。
3. handler 内部必须捕获所有异常。让 handler 抛出未捕获异常会直接中断整个 query() 调用，调试很麻烦。统一用 is_error: True 把错误作为数据回传，让 Claude 处理。
4. 多工具用通配符 mcp__server__* 放行。一个 server 里工具多了后逐个列 allowed_tools 很繁琐，通配符更好维护。如果只想暴露其中几个，逐个列出；想全部放行，用通配符。
5. 工具多到几十个时，考虑 Tool Search。每个工具的 schema 都消耗 context window，超过 20 个工具时建议用 Tool Search 懒加载机制（#8 已介绍过 MCP 集成中的同类机制）按需加载。