MCP Tool Calling 高级模式:构建智能工具调用链
探索 Model Context Protocol 中 Tool Calling 的高级设计模式,包括链式调用、条件路由、错误恢复和并行执行。
Tool Calling 是 MCP 的核心能力——LLM 通过调用工具来获取信息、执行操作。但简单的工具调用只是起点,真正的价值在于如何设计智能的调用模式,让 Agent 能够高效地完成复杂任务。
基础调用模式回顾
在进入高级模式之前,先回顾基础的工具调用流程:
用户请求 → LLM 分析 → 选择工具 → 执行 → 返回结果 → LLM 继续推理
这个单次调用模式足以处理简单任务,但面对复杂场景时显得力不从心。
链式调用模式
链式调用是指多个工具按顺序执行,前一个工具的输出作为后一个工具的输入。
设计思路
将复杂任务分解为多个步骤,每个步骤由一个专门的工具完成。LLM 负责理解任务并编排调用顺序。
// 数据处理链:获取数据 → 清洗 → 分析 → 输出
server.tool('fetch_data', '从数据源获取原始数据', {
source: z.string(),
}, async ({ source }) => {
const raw = await fetchDataFromSource(source);
return { content: [{ type: 'text', text: JSON.stringify(raw) }] };
});
server.tool('clean_data', '清洗和标准化数据', {
data: z.string().describe('JSON 格式的原始数据'),
}, async ({ data }) => {
const parsed = JSON.parse(data);
const cleaned = removeDuplicates(normalize(parsed));
return { content: [{ type: 'text', text: JSON.stringify(cleaned) }] };
});
server.tool('analyze_data', '分析数据并生成洞察', {
data: z.string().describe('JSON 格式的清洗后数据'),
}, async ({ data }) => {
const parsed = JSON.parse(data);
const insights = generateInsights(parsed);
return { content: [{ type: 'text', text: insights }] };
});
LLM 会自动编排调用顺序:fetch_data → clean_data → analyze_data。
内置链式调用
你也可以在 Server 端直接实现链式调用,减少 LLM 的推理负担:
server.tool('process_pipeline', '执行完整的数据处理流水线', {
source: z.string(),
operations: z.array(z.enum(['clean', 'analyze', 'export'])),
}, async ({ source, operations }) => {
let data = await fetchDataFromSource(source);
for (const op of operations) {
switch (op) {
case 'clean':
data = cleanData(data);
break;
case 'analyze':
data = analyzeData(data);
break;
case 'export':
data = exportData(data);
break;
}
}
return { content: [{ type: 'text', text: JSON.stringify(data) }] };
});
条件路由模式
根据不同的输入条件,使用不同的处理逻辑。
基于数据类型的路由
server.tool('smart_query', '智能查询:根据数据源类型自动选择查询方式', {
query: z.string(),
sourceType: z.enum(['database', 'api', 'file']),
}, async ({ query, sourceType }) => {
let result: string;
switch (sourceType) {
case 'database':
result = await queryDatabase(query);
break;
case 'api':
result = await callExternalAPI(query);
break;
case 'file':
result = await searchFiles(query);
break;
}
return { content: [{ type: 'text', text: result }] };
});
基于置信度的路由
当 LLM 不确定使用哪个工具时,提供一个”智能路由”工具:
server.tool('auto_search', '自动选择最佳搜索策略', {
query: z.string(),
context: z.string().optional().describe('搜索上下文'),
}, async ({ query, context }) => {
// 分析查询意图
const intent = classifyIntent(query);
switch (intent.type) {
case 'factual':
return await webSearch(query);
case 'code':
return await codeSearch(query);
case 'document':
return await documentSearch(query, context);
default:
return await hybridSearch(query);
}
});
并行执行模式
当多个工具调用之间没有依赖关系时,可以并行执行以提高效率。
LLM 原生并行
Claude 等 LLM 可以在一次响应中返回多个 tool_use 块,Client 可以并行执行:
async function handleToolCalls(toolCalls: ToolCall[]): Promise<ToolResult[]> {
// 并行执行所有工具调用
const results = await Promise.all(
toolCalls.map(async (call) => {
const result = await executeTool(call.name, call.input);
return {
type: 'tool_result',
tool_use_id: call.id,
content: result,
};
})
);
return results;
}
批量操作工具
设计支持批量操作的工具,减少调用次数:
server.tool('batch_read', '批量读取多个文件', {
filePaths: z.array(z.string()).describe('文件路径列表'),
}, async ({ filePaths }) => {
const results = await Promise.allSettled(
filePaths.map(async (fp) => {
const content = await fs.readFile(fp, 'utf-8');
return { path: fp, content };
})
);
const output = results.map((r, i) => {
if (r.status === 'fulfilled') {
return `### ${filePaths[i]}\n${r.value.content}`;
}
return `### ${filePaths[i]}\n错误:${r.reason}`;
});
return { content: [{ type: 'text', text: output.join('\n\n') }] };
});
错误恢复模式
工具执行可能失败,良好的错误恢复机制能提升 Agent 的鲁棒性。
自动重试
async function callWithRetry(
fn: () => Promise<string>,
maxRetries: number = 3
): Promise<string> {
for (let i = 0; i < maxRetries; i++) {
try {
return await fn();
} catch (error) {
if (i === maxRetries - 1) throw error;
await new Promise(r => setTimeout(r, 1000 * Math.pow(2, i)));
}
}
throw new Error('Unreachable');
}
server.tool('reliable_fetch', '带自动重试的数据获取', {
url: z.string().url(),
}, async ({ url }) => {
const result = await callWithRetry(() => fetch(url).then(r => r.text()));
return { content: [{ type: 'text', text: result }] };
});
降级策略
当首选工具失败时,自动降级到备选方案:
server.tool('search_with_fallback', '带降级的搜索', {
query: z.string(),
}, async ({ query }) => {
try {
// 首选:精确搜索
const exact = await exactSearch(query);
if (exact.length > 0) {
return { content: [{ type: 'text', text: formatResults(exact) }] };
}
} catch {}
try {
// 降级:模糊搜索
const fuzzy = await fuzzySearch(query);
return { content: [{ type: 'text', text: formatResults(fuzzy) }] };
} catch {}
// 最终降级:网络搜索
const web = await webSearch(query);
return { content: [{ type: 'text', text: web }] };
});
上下文感知模式
工具根据对话上下文动态调整行为。
记忆工具
const conversationMemory = new Map<string, any[]>();
server.tool('remember', '存储信息到对话记忆', {
key: z.string(),
value: z.string(),
sessionId: z.string(),
}, async ({ key, value, sessionId }) => {
if (!conversationMemory.has(sessionId)) {
conversationMemory.set(sessionId, []);
}
conversationMemory.get(sessionId)!.push({ key, value, timestamp: Date.now() });
return { content: [{ type: 'text', text: `已记住:${key}` }] };
});
server.tool('recall', '从对话记忆中检索信息', {
query: z.string(),
sessionId: z.string(),
}, async ({ query, sessionId }) => {
const memory = conversationMemory.get(sessionId) || [];
const relevant = memory.filter(m =>
m.key.includes(query) || m.value.includes(query)
);
return {
content: [{ type: 'text', text: JSON.stringify(relevant) }],
};
});
常见问题(FAQ)
LLM 如何知道工具之间的依赖关系?
通过工具的描述和参数定义。清晰的描述能帮助 LLM 理解工具的输入输出关系,从而正确编排调用顺序。
并行调用会导致数据竞争吗?
MCP 协议本身不处理并发控制。如果多个工具可能修改同一资源,需要在 Server 端实现适当的锁机制。
如何限制工具调用的次数?
在 Client 端设置最大迭代次数,当 LLM 连续调用工具超过阈值时强制终止循环。
总结
高级 Tool Calling 模式让 MCP 的能力得到了质的提升。链式调用实现复杂流程,条件路由处理多样场景,并行执行提高效率,错误恢复保证可靠性。设计这些模式的关键在于:让工具定义本身引导 LLM 做出正确的调用决策。