引言
工具调用(Tool Calling) 是MCP协议最核心的功能。它标准化了AI模型如何发现、选择和调用外部工具,是连接LLM与现实世界的桥梁。
本文将深入解析:
- 工具定义规范与Schema格式
tools/list工具发现流程tools/call调用执行流程- 参数验证与结果处理
- 错误处理与重试策略
- 高级特性:流式响应与进度通知
工具定义规范
完整的工具Schema
MCP中的每个工具都通过一个标准化的JSON Schema描述:
{
"tools": [
{
"name": "search_codebase",
"description": "在代码库中搜索特定模式,支持正则和文件类型过滤",
"inputSchema": {
"type": "object",
"properties": {
"pattern": {
"type": "string",
"description": "搜索模式,支持正则表达式"
},
"fileType": {
"type": "string",
"description": "文件类型过滤,如 .ts, .py, .md",
"enum": [".ts", ".py", ".js", ".md", ".rs", ".go"]
},
"caseSensitive": {
"type": "boolean",
"description": "是否区分大小写",
"default": false
},
"maxResults": {
"type": "integer",
"description": "最大返回结果数",
"minimum": 1,
"maximum": 100,
"default": 20
}
},
"required": ["pattern"]
}
}
]
}
| 字段 | 类型 | 必填 | 说明 |
|---|---|---|---|
name | string | 是 | 工具唯一标识符,在同Server中必须唯一 |
description | string | 推荐 | 工具功能的详细描述,帮助LLM理解何时使用 |
inputSchema | object | 推荐 | JSON Schema描述的输入参数定义 |
工具发现流程:tools/list
请求格式
Client向Server请求可用工具列表:
{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/list"
}
支持分页查询:
{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/list",
"params": {
"cursor": "next_page_token"
}
}
响应格式
{
"jsonrpc": "2.0",
"id": 1,
"result": {
"tools": [
{
"name": "search_codebase",
"description": "在代码库中搜索代码",
"inputSchema": {
"type": "object",
"properties": {
"pattern": { "type": "string" },
"fileType": { "type": "string" }
},
"required": ["pattern"]
}
},
{
"name": "get_file_content",
"description": "读取文件内容",
"inputSchema": {
"type": "object",
"properties": {
"path": { "type": "string" },
"startLine": { "type": "integer" },
"endLine": { "type": "integer" }
},
"required": ["path"]
}
}
],
"nextCursor": null
}
}
Server端实现
// TypeScript: tools/list处理器
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import {
ListToolsRequestSchema,
} from "@modelcontextprotocol/sdk/types.js";
const server = new Server(
{ name: "codebase-server", version: "1.0.0" },
{ capabilities: { tools: {} } }
);
server.setRequestHandler(ListToolsRequestSchema, async (request) => {
const { cursor } = request.params ?? {};
// 支持分页
if (cursor === "page2") {
return {
tools: [/* 第二页工具 */],
nextCursor: null,
};
}
return {
tools: [
{
name: "search_codebase",
description: "在代码库中搜索模式",
inputSchema: {
type: "object",
properties: {
pattern: { type: "string", description: "搜索模式" },
fileType: { type: "string", description: "文件类型过滤" }
},
required: ["pattern"]
}
},
{
name: "get_file_content",
description: "读取文件内容",
inputSchema: {
type: "object",
properties: {
path: { type: "string", description: "文件路径" }
},
required: ["path"]
}
}
],
nextCursor: null
};
});
工具调用流程:tools/call
完整调用流程
LLM (决策层) MCP Client MCP Server 外部系统
│ │ │ │
│ "搜索MCP代码" │ │ │
│───────────────────►│ │ │
│ │ │ │
│ │── tools/call ─────►│ │
│ │ search_codebase │ │
│ │ {pattern: "MCP"} │ │
│ │ │ │
│ │ │── 执行搜索 ──►│
│ │ │◄── 返回结果 ──│
│ │ │ │
│ │◄─ tool result ─────│ │
│ │ {content: [...]} │ │
│ │ │ │
│◄── 返回结果 ──────│ │ │
│ │ │ │
请求格式
{
"jsonrpc": "2.0",
"id": 2,
"method": "tools/call",
"params": {
"name": "search_codebase",
"arguments": {
"pattern": "MCP协议",
"fileType": ".md",
"caseSensitive": false,
"maxResults": 10
}
}
}
成功响应
{
"jsonrpc": "2.0",
"id": 2,
"result": {
"content": [
{
"type": "text",
"text": "找到 3 个匹配结果:\n\n1. docs/mcp-intro.md:15 - MCP协议定义\n2. docs/mcp-architecture.md:42 - MCP架构\n3. examples/mcp-server.md:1 - MCP Server示例"
}
],
"isError": false
}
}
结果内容类型
MCP支持多种内容类型返回:
| 类型 | 用途 | 示例 |
|---|---|---|
text | 文本内容 | 搜索结果、查询结果 |
image | 图片数据 | 截图、图表(需Base64编码) |
audio | 音频数据 | 语音识别结果(需Base64编码) |
resource | 资源引用 | 文件内容引用 |
{
"content": [
{
"type": "text",
"text": "查询结果摘要如下:"
},
{
"type": "image",
"data": "iVBORw0KGgo...",
"mimeType": "image/png"
},
{
"type": "text",
"text": "以上是数据可视化结果"
}
],
"isError": false
}
Server端实现
server.setRequestHandler(CallToolRequestSchema, async (request) => {
const { name, arguments: args } = request.params;
switch (name) {
case "search_codebase": {
// 参数验证
if (!args?.pattern || typeof args.pattern !== "string") {
throw new Error("缺少必填参数: pattern");
}
try {
const results = await searchCodebase(args.pattern, {
fileType: args.fileType,
caseSensitive: args.caseSensitive,
maxResults: args.maxResults,
});
return {
content: [
{
type: "text",
text: formatSearchResults(results),
},
],
isError: false,
};
} catch (error) {
return {
content: [{ type: "text", text: `搜索失败: ${error.message}` }],
isError: true,
};
}
}
case "get_file_content": {
// 实现文件读取逻辑
// ...
}
default:
throw new Error(`未知工具: ${name}`);
}
});
参数验证
Zod Schema验证(TypeScript)
import { z } from "zod";
// 定义参数验证Schema
const SearchArgsSchema = z.object({
pattern: z.string().min(1, "搜索模式不能为空"),
fileType: z.enum([".ts", ".py", ".js", ".md"]).optional(),
caseSensitive: z.boolean().default(false),
maxResults: z.number().int().min(1).max(100).default(20),
});
server.setRequestHandler(CallToolRequestSchema, async (request) => {
const { name, arguments: args } = request.params;
if (name === "search_codebase") {
// 使用Zod进行类型安全和安全的参数验证
const validArgs = SearchArgsSchema.parse(args);
// validArgs现在有正确的类型和默认值
return await executeSearch(validArgs);
}
});
Pydantic验证(Python)
from pydantic import BaseModel, Field
from typing import Optional
class SearchArgs(BaseModel):
pattern: str = Field(..., min_length=1, description="搜索模式")
file_type: Optional[str] = Field(None, description="文件类型过滤")
case_sensitive: bool = Field(False, description="是否区分大小写")
max_results: int = Field(20, ge=1, le=100, description="最大结果数")
def handle_tool_call(name: str, arguments: dict) -> dict:
if name == "search_codebase":
args = SearchArgs(**arguments)
# args 已经过完整验证
results = execute_search(args)
return {
"content": [{"type": "text", "text": str(results)}],
"isError": False,
}
错误处理
标准错误响应
# 工具执行中的各类错误处理
def handle_tool_error(name: str, error: Exception) -> dict:
if isinstance(error, ValidationError):
return {
"content": [
{"type": "text",
"text": f"参数验证失败: {error}"},
],
"isError": True,
}
elif isinstance(error, TimeoutError):
return {
"content": [
{"type": "text",
"text": f"工具 {name} 执行超时"},
],
"isError": True,
}
elif isinstance(error, PermissionError):
return {
"content": [
{"type": "text",
"text": f"权限不足: {error}"},
],
"isError": True,
}
else:
return {
"content": [
{"type": "text",
"text": f"工具执行失败: {error}"},
],
"isError": True,
}
高级特性
工具列表变更通知
当Server的工具列表发生变化时,可以主动通知Client:
// Server端:工具列表变更时发送通知
server.notification({
method: "notifications/tools/list_changed",
});
进度通知
长时间运行的工具调用可以发送进度更新:
// 包含进度令牌的调用
{
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
"params": {
"name": "process_large_file",
"arguments": { "path": "/data/large.csv" },
"_meta": {
"progressToken": "progress-1"
}
}
}
// 服务端发送进度通知
{
"jsonrpc": "2.0",
"method": "notifications/progress",
"params": {
"progressToken": "progress-1",
"progress": 50,
"total": 100,
"message": "正在处理第 5000/10000 行..."
}
}
工具设计最佳实践
工具设计原则
| 原则 | 说明 | 反例 |
|---|---|---|
| 单一职责 | 每个工具只做一件事 | 一个”系统工具”又读文件又发消息 |
| 清晰的description | 帮助LLM判断何时使用 | ”处理数据” vs “将CSV数据转换为JSON格式” |
| 合理的默认值 | 减少LLM需要提供的参数 | 搜索结果数默认20 |
| 参数约束 | 明确的类型、范围、枚举 | 防止LLM生成非法输入 |
工具命名规范
✅ 好的命名: search_codebase, send_email, generate_report
❌ 差的命名: do_stuff, process, utils_execute
复杂的description写法
// ✅ 好的description - 帮助LLM理解
{
"name": "calculate_complex",
"description": "执行复杂的数学计算。支持加减乘除(+ - * /)、" +
"幂运算(**)、三角函数(sin, cos, tan)、对数(log, ln)、" +
"开平方(sqrt)。适用于需要精确数值计算的场景。" +
"不适合简单计算(加减乘除在对话中直接回答)。",
// ...
}
与其他协议对比
| 特性 | MCP tools | OpenAI Function Calling | Anthropic Tool Use |
|---|---|---|---|
| 标准化 | 开放标准 | 厂商绑定 | 厂商绑定 |
| 工具发现 | tools/list | 静态注册 | 静态注册 |
| Schema格式 | JSON Schema | JSON Schema | JSON Schema |
| 结果格式 | 标准化content | 自由格式 | 自由格式 |
| 流式支持 | 进度通知 | SSE | 流式 |
总结
MCP的工具调用协议标准化了AI模型与外部工具的交互:
| 特性 | 说明 | 优势 |
|---|---|---|
| tools/list | 动态工具发现 | Client无需预知工具列表 |
| tools/call | 标准化调用流程 | 统一的结果和错误格式 |
| inputSchema | JSON Schema验证 | 类型安全和参数校验 |
| 进度通知 | 长时间操作反馈 | 提升用户体验 |
下一步学习建议:
本文最后更新于 2024-07-02。