Claude Code官方文档Agent SDKMCP自定义工具

Claude Agent SDK 自定义工具 - in-process MCP server 完整指南

使用 Claude Agent SDK 的进程内 MCP 服务器定义自定义工具，让 Claude 调用你的函数、访问 API 并执行特定领域的操作。

· 阅读约 26 分钟

为 Claude 提供自定义工具

使用 Claude Agent SDK 的进程内 MCP 服务器定义自定义工具，以便 Claude 可以调用您的函数、访问您的 API 并执行特定领域的操作。

自定义工具通过让您定义 Claude 在对话期间可以调用的自己的函数来扩展 Agent SDK。使用 SDK 的进程内 MCP 服务器，您可以让 Claude 访问数据库、外部 API、特定领域的逻辑或应用程序需要的任何其他功能。

本指南涵盖如何使用输入架构和处理程序定义工具、将它们捆绑到 MCP 服务器中、将它们传递给 query，以及控制 Claude 可以访问哪些工具。它还涵盖错误处理、工具注释和返回非文本内容（如图像）。

快速参考

如果您想…执行此操作
定义工具使用 @tool（Python）或 tool()（TypeScript），包含名称、描述、架构和处理程序。
向 Claude 注册工具在 create_sdk_mcp_server / createSdkMcpServer 中包装并传递给 query() 中的 mcpServers。
预先批准工具添加到您的允许工具列表。
从 Claude 的上下文中删除内置工具传递仅列出您想要的内置工具的 tools 数组。
让 Claude 并行调用工具在没有副作用的工具上设置 readOnlyHint: true。
处理错误而不停止循环返回 isError: true 而不是抛出异常。
返回图像或文件在内容数组中使用 image 或 resource 块。
返回机器可读的 JSON 结果在结果上设置 structuredContent。
扩展到许多工具使用工具搜索按需加载工具。

创建自定义工具

工具由四个部分定义，作为参数传递给 TypeScript 中的 tool() 助手或 Python 中的 @tool 装饰器：

  • 名称： Claude 用来调用工具的唯一标识符。
  • 描述： 工具的功能。Claude 读取此内容以决定何时调用它。
  • 输入架构： Claude 必须提供的参数。在 TypeScript 中，这始终是 Zod 架构，处理程序的 args 会自动从中获得类型。在 Python 中，这是一个将名称映射到类型的字典，如 {"latitude": float}，SDK 会为您将其转换为 JSON Schema。Python 装饰器还接受完整的 JSON Schema 字典，当您需要枚举、范围、可选字段或嵌套对象时。
  • 处理程序： 当 Claude 调用工具时运行的异步函数。它接收验证的参数，必须返回一个对象，包含：
    • content（必需）：结果块的数组，每个块的 type 为 "text"、"image" 或 "resource"。
    • structuredContent（可选）：保存结果作为机器可读数据的 JSON 对象，与 content 一起返回。
    • isError（可选）：设置为 true 以表示工具失败，以便 Claude 可以对其做出反应。

定义工具后，使用 createSdkMcpServer（TypeScript）或 create_sdk_mcp_server（Python）将其包装在服务器中。服务器在应用程序内进程内运行，而不是作为单独的进程。

天气工具示例

此示例定义了一个 get_temperature 工具并将其包装在 MCP 服务器中。它仅设置工具；要将其传递给 query 并运行它，请参阅下面的调用自定义工具。

Python：

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


# Define a tool: name, description, input schema, handler
@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 a content array - Claude sees this as the tool result
    return {
        "content": [
            {
                "type": "text",
                "text": f"Temperature: {data['current']['temperature_2m']}°F",
            }
        ]
    }


# Wrap the tool in an in-process MCP server
weather_server = create_sdk_mcp_server(
    name="weather",
    version="1.0.0",
    tools=[get_temperature],
)

TypeScript：

import { tool, createSdkMcpServer } from "@anthropic-ai/claude-agent-sdk";
import { z } from "zod";

// Define a tool: name, description, input schema, handler
const getTemperature = tool(
  "get_temperature",
  "Get the current temperature at a location",
  {
    latitude: z.number().describe("Latitude coordinate"),
    longitude: z.number().describe("Longitude coordinate")
  },
  async (args) => {
    const response = await fetch(
      `https://api.open-meteo.com/v1/forecast?latitude=${args.latitude}&longitude=${args.longitude}&current=temperature_2m&temperature_unit=fahrenheit`
    );
    const data: any = await response.json();

    return {
      content: [{ type: "text", text: `Temperature: ${data.current.temperature_2m}°F` }]
    };
  }
);

// Wrap the tool in an in-process MCP server
const weatherServer = createSdkMcpServer({
  name: "weather",
  version: "1.0.0",
  tools: [getTemperature]
});

💡 要使参数可选：在 TypeScript 中，向 Zod 字段添加 .default()。在 Python 中，字典架构将每个键视为必需的，因此将参数从架构中省略，在描述字符串中提及它，并在处理程序中使用 args.get() 读取它。

调用自定义工具

通过 mcpServers 选项将您创建的 MCP 服务器传递给 query。mcpServers 中的键成为每个工具的完全限定名称中的 {server_name} 段：mcp__{server_name}__{tool_name}。在 allowedTools 中列出该名称，以便工具运行而无需权限提示。

Python：

import asyncio
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)


asyncio.run(main())

TypeScript：

import { query } from "@anthropic-ai/claude-agent-sdk";

for await (const message of query({
  prompt: "What's the temperature in San Francisco?",
  options: {
    mcpServers: { weather: weatherServer },
    allowedTools: ["mcp__weather__get_temperature"]
  }
})) {
  if (message.type === "result" && message.subtype === "success") {
    console.log(message.result);
  }
}

添加更多工具

一个服务器在其 tools 数组中列出的工具数量不限。如果有多个工具在一个服务器上，您可以在 allowedTools 中单独列出每个工具，或使用通配符 mcp__weather__* 来覆盖服务器公开的每个工具。

Python：

# Define a second tool for the same server
@tool(
    "get_precipitation_chance",
    "Get the hourly precipitation probability for a location. "
    "Optionally pass 'hours' (1-24) to control how many hours to return.",
    {"latitude": float, "longitude": float},
)
async def get_precipitation_chance(args: dict[str, Any]) -> dict[str, Any]:
    # 'hours' isn't in the schema - read it with .get() to make it optional
    hours = args.get("hours", 12)
    async with httpx.AsyncClient() as client:
        response = await client.get(
            "https://api.open-meteo.com/v1/forecast",
            params={
                "latitude": args["latitude"],
                "longitude": args["longitude"],
                "hourly": "precipitation_probability",
                "forecast_days": 1,
            },
        )
        data = response.json()
    chances = data["hourly"]["precipitation_probability"][:hours]

    return {
        "content": [
            {
                "type": "text",
                "text": f"Next {hours} hours: {'%, '.join(map(str, chances))}%",
            }
        ]
    }


# Rebuild the server with both tools in the array
weather_server = create_sdk_mcp_server(
    name="weather",
    version="1.0.0",
    tools=[get_temperature, get_precipitation_chance],
)

TypeScript：

const getPrecipitationChance = tool(
  "get_precipitation_chance",
  "Get the hourly precipitation probability for a location",
  {
    latitude: z.number(),
    longitude: z.number(),
    hours: z
      .number()
      .int()
      .min(1)
      .max(24)
      .default(12)
      .describe("How many hours of forecast to return")
  },
  async (args) => {
    const response = await fetch(
      `https://api.open-meteo.com/v1/forecast?latitude=${args.latitude}&longitude=${args.longitude}&hourly=precipitation_probability&forecast_days=1`
    );
    const data: any = await response.json();
    const chances = data.hourly.precipitation_probability.slice(0, args.hours);

    return {
      content: [{ type: "text", text: `Next ${args.hours} hours: ${chances.join("%, ")}%` }]
    };
  }
);

const weatherServer = createSdkMcpServer({
  name: "weather",
  version: "1.0.0",
  tools: [getTemperature, getPrecipitationChance]
});

此数组中的每个工具在每个回合都会消耗上下文窗口空间。如果您定义了数十个工具，请参阅工具搜索以按需加载它们。

添加工具注释

工具注释是描述工具行为方式的可选元数据。在 TypeScript 中作为 tool() 助手的第五个参数传递，或在 Python 中通过 @tool 装饰器的 annotations 关键字参数传递。所有提示字段都是布尔值。

字段默认值含义
readOnlyHintfalse工具不修改其环境。控制工具是否可以与其他只读工具并行调用。
destructiveHinttrue工具可能执行破坏性更新。仅供参考。
idempotentHintfalse使用相同参数的重复调用没有额外效果。仅供参考。
openWorldHinttrue工具到达流程外的系统。仅供参考。

注释是元数据，不是强制执行。标记为 readOnlyHint: true 的工具如果处理程序这样做，仍然可以写入磁盘。保持注释与处理程序准确。

Python：

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
    ),  # Lets Claude batch this with other read-only calls
)
async def get_temperature(args):
    return {"content": [{"type": "text", "text": "..."}]}

TypeScript：

tool(
  "get_temperature",
  "Get the current temperature at a location",
  { latitude: z.number(), longitude: z.number() },
  async (args) => ({ content: [{ type: "text", text: `...` }] }),
  { annotations: { readOnlyHint: true } } // Lets Claude batch this with other read-only calls
);

控制工具访问

天气工具示例注册了一个服务器并在 allowedTools 中列出了工具。本部分涵盖工具名称的构造方式以及当您有多个工具或想要限制内置工具时如何限制访问。

工具名称格式

当 MCP 工具暴露给 Claude 时，它们的名称遵循特定格式：

  • 模式：mcp__{server_name}__{tool_name}
  • 示例：服务器 weather 中名为 get_temperature 的工具变成 mcp__weather__get_temperature

配置允许的工具

tools 选项和允许/不允许列表在不同的层上运行。tools 控制哪些内置工具出现在 Claude 的上下文中。允许和不允许的工具列表控制 Claude 尝试调用它们后是否批准或拒绝调用。

选项层效果
tools: ["Read", "Grep"]可用性仅列出的内置工具在 Claude 的上下文中。未列出的内置工具被删除。MCP 工具不受影响。
tools: []可用性所有内置工具都被删除。Claude 只能使用您的 MCP 工具。
允许的工具权限列出的工具运行而无需权限提示。未列出的工具保持可用；调用通过权限流进行。
不允许的工具权限对列出的工具的每次调用都被拒绝。工具保留在 Claude 的上下文中，因此 Claude 可能仍会在调用被拒绝之前尝试它。

要限制 Claude 可以使用哪些内置工具，优先使用 tools 而不是不允许的工具。从 tools 中省略工具会将其从上下文中删除，以便 Claude 永远不会尝试它；在 disallowedTools 中列出它（Python：disallowed_tools）会阻止调用但保留工具可见，因此 Claude 可能会浪费一个回合尝试它。

处理错误

您的处理程序报告错误的方式决定了代理循环是继续还是停止：

发生的情况结果
处理程序抛出未捕获的异常代理循环停止。Claude 永远看不到错误，query 调用失败。
处理程序捕获错误并返回 isError: true（TS）/ "is_error": True（Python）代理循环继续。Claude 将错误视为数据，可以重试、尝试不同的工具或解释失败。

下面的示例在处理程序内部捕获两种失败，而不是让它们抛出。

Python：

import json
import httpx
from typing import Any


@tool(
    "fetch_data",
    "Fetch data from an API",
    {"endpoint": str},  # Simple schema
)
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,
                }

            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,
        }

TypeScript：

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}`
            }
          ],
          isError: true
        };
      }

      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 instanceof Error ? error.message : String(error)}`
          }
        ],
        isError: true
      };
    }
  }
);

返回图像和资源

工具结果中的 content 数组接受 text、image 和 resource 块。您可以在同一响应中混合它们。

图像

图像块以 base64 编码的方式内联携带图像字节。没有 URL 字段。要返回位于 URL 的图像，在处理程序中获取它，读取响应字节，并在返回之前进行 base64 编码。结果作为视觉输入处理。

字段类型注释
type"image"
datastringBase64 编码的字节。仅原始 base64，没有 data:image/...;base64, 前缀
mimeTypestring必需。例如 image/png、image/jpeg、image/webp、image/gif

Python：

import base64
import httpx


@tool("fetch_image", "Fetch an image from a URL and return it to Claude", {"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"),
            }
        ]
    }

TypeScript：

tool(
  "fetch_image",
  "Fetch an image from a URL and return it to Claude",
  {
    url: z.string().url()
  },
  async (args) => {
    const response = await fetch(args.url);
    const buffer = Buffer.from(await response.arrayBuffer());
    const mimeType = response.headers.get("content-type") ?? "image/png";

    return {
      content: [
        {
          type: "image",
          data: buffer.toString("base64"),
          mimeType
        }
      ]
    };
  }
);

资源

资源块嵌入由 URI 标识的内容片段。URI 是 Claude 引用的标签；实际内容位于块的 text 或 blob 字段中。

字段类型注释
type"resource"
resource.uristring内容的标识符。任何 URI 方案
resource.textstring内容，如果是文本。提供此项或 blob，不能两者都提供
resource.blobstring内容 base64 编码，如果是二进制
resource.mimeTypestring可选

TypeScript：

return {
  content: [
    {
      type: "resource",
      resource: {
        uri: "file:///tmp/report.md",
        mimeType: "text/markdown",
        text: "# Report\n..."
      }
    }
  ]
};

Python：

return {
    "content": [
        {
            "type": "resource",
            "resource": {
                "uri": "file:///tmp/report.md",
                "mimeType": "text/markdown",
                "text": "# Report\n...",
            },
        }
    ]
}

返回结构化数据

structuredContent 是结果上的可选 JSON 对象，与 content 数组分开。使用它返回原始值，Claude 可以将其作为精确字段读取，而不是从文本字符串或图像中解析它们。

当设置 structuredContent 时，Claude 接收 JSON 加上来自 content 的任何图像或资源块。来自 content 的文本块不被转发，因为假设它们复制结构化数据。

return {
  content: [
    {
      type: "image",
      data: chartPngBuffer.toString("base64"),
      mimeType: "image/png"
    }
  ],
  structuredContent: {
    series: "temperature_2m",
    unit: "fahrenheit",
    points: [62.1, 63.4, 65.0, 64.2]
  }
};

ℹ️ Python @tool 装饰器仅从处理程序的返回字典转发 content 和 is_error。要从 Python 返回 structuredContent，请运行独立 MCP 服务器而不是进程内 SDK 服务器。

后续步骤

自定义工具在标准接口中包装异步函数。您可以在同一服务器中混合本页上的模式：单个服务器可以在彼此旁边保存数据库工具、API 网关工具和图像渲染器。

从这里：

  • 如果您的服务器增长到数十个工具，请参阅工具搜索以延迟加载它们，直到 Claude 需要它们。
  • 要连接到外部 MCP 服务器（文件系统、GitHub、Slack）而不是构建自己的，请参阅连接 MCP 服务器。
  • 要控制哪些工具自动运行与需要批准，请参阅配置权限。

相关文档

本文翻译自 Anthropic Claude Code 官方文档，最近一次同步：2025-05-01。