大模型工具调用与 MCP：格式、并行与安全边界

1. 引言大模型LLM本身是“文本生成器”无法直接执行外部操作比如查询数据库、调用 API、读写文件或发送邮件。工具调用Function Calling / Tool Use让模型在对话中声明“我需要调用某个工具”由外部系统真正执行再把结果回传给模型继续推理。MCPModel Context Protocol则把“工具、资源、提示词”统一成一套标准化协议让模型可以跨应用复用同一套工具生态。本文从格式、并行调用和安全边界三个维度展开并给出可运行的代码实战。2. 工具调用的核心格式不同厂商对工具调用的消息格式略有差异但核心思路一致模型输出一个结构化的“工具调用请求”而不是直接执行代码。以 OpenAI 风格为例工具调用通常包含工具名称、参数和调用 ID。一个典型的工具调用请求如下{ role: assistant, content: null, tool_calls: [ { id: call_abc123, type: function, function: { name: get_weather, arguments: {\city\: \北京\, \date\: \2026-08-09\} } } ] }外部系统执行后把结果以“工具消息”回传给模型{ role: tool, tool_call_id: call_abc123, content: {\temperature\: 32, \condition\: \晴\} }模型拿到工具结果后继续生成面向用户的最终回答。这个“请求-执行-回传-续答”的循环就是工具调用的基本工作流。3. 工具定义与参数约束为了让模型正确调用工具开发者需要提供工具的结构化定义包括名称、描述和参数 JSON Schema。描述越清晰模型选错工具的概率越低。tools [ { type: function, function: { name: get_weather, description: 查询指定城市在指定日期的天气情况, parameters: { type: object, properties: { city: {type: string, description: 城市名称如北京、上海}, date: {type: string, description: 日期格式 YYYY-MM-DD} }, required: [city, date] } } } ]参数 Schema 中应尽量使用 enum、format 等约束字段减少模型生成非法参数的概率。例如日期字段可以补充 pattern 校验。4. 并行工具调用当一次回答需要调用多个相互独立的工具时模型可以在一次响应中返回多个 tool_calls由外部系统并行执行从而显著降低延迟。并行调用示例{ role: assistant, content: null, tool_calls: [ { id: call_1, function: {name: get_weather, arguments: {\city\: \北京\}} }, { id: call_2, function: {name: get_weather, arguments: {\city\: \上海\}} }, { id: call_3, function: {name: get_stock_price, arguments: {\symbol\: \AAPL\}} } ] }外部系统应使用并发方式执行这些调用例如 Python 的 asyncio.gather 或线程池。需要注意并行调用只适用于相互之间没有依赖关系的工具如果工具 B 的入参依赖工具 A 的输出则必须串行执行。5. 代码实战完整工具调用循环下面给出一个完整的 Python 示例演示“模型声明调用-外部执行-结果回传-模型续答”的闭环。示例使用 OpenAI SDK 风格但核心逻辑适用于大多数兼容接口。import json from openai import OpenAI client OpenAI() def get_weather(city: str) - str: 模拟天气查询工具 data {北京: 32, 上海: 28, 广州: 30} return json.dumps({city: city, temperature: data.get(city, 25)}) tools [ { type: function, function: { name: get_weather, description: 查询指定城市的天气温度, parameters: { type: object, properties: { city: {type: string, description: 城市名称} }, required: [city] } } } ] messages [{role: user, content: 北京和上海今天多少度}] 第一轮模型可能返回工具调用请求 response client.chat.completions.create( modelgpt-4o, messagesmessages, toolstools, ) assistant_msg response.choices[0].message messages.append(assistant_msg) 检查是否有工具调用 if assistant_msg.tool_calls: for tc in assistant_msg.tool_calls: args json.loads(tc.function.arguments) result get_weather(args[city]) messages.append({ role: tool, tool_call_id: tc.id, content: result, }) 第二轮模型基于工具结果生成最终回答 final_response client.chat.completions.create( modelgpt-4o, messagesmessages, toolstools, ) print(final_response.choices[0].message.content)这段代码的关键点在于assistant 消息必须原样追加回 messages工具结果必须通过 tool_call_id 与对应的调用请求关联否则模型无法正确理解哪个结果对应哪个调用。6. 并行调用实战asyncio 实现当模型一次返回多个 tool_calls 时可以使用 asyncio 并发执行。下面给出一个可运行的并行示例。import asyncio import json from openai import AsyncOpenAI client AsyncOpenAI() async def call_tool(name: str, arguments: str) - str: 根据工具名分发执行 args json.loads(arguments) if name get_weather: data {北京: 32, 上海: 28} return json.dumps({city: args[city], temperature: data.get(args[city], 25)}) if name get_stock: return json.dumps({symbol: args[symbol], price: 188.5}) return json.dumps({error: unknown tool}) async def main(): messages [{role: user, content: 查一下北京天气和 AAPL 股价}] tools [ {type: function, function: {name: get_weather, description: 查天气, parameters: {type: object, properties: {city: {type: string}}, required: [city]}}}, {type: function, function: {name: get_stock, description: 查股价, parameters: {type: object, properties: {symbol: {type: string}}, required: [symbol]}}}, ] resp await client.chat.completions.create(modelgpt-4o, messagesmessages, toolstools) assistant_msg resp.choices[0].message messages.append(assistant_msg) if assistant_msg.tool_calls: # 并发执行所有工具调用 results await asyncio.gather(*[ call_tool(tc.function.name, tc.function.arguments) for tc in assistant_msg.tool_calls ]) for tc, result in zip(assistant_msg.tool_calls, results): messages.append({role: tool, tool_call_id: tc.id, content: result}) final await client.chat.completions.create(modelgpt-4o, messagesmessages, toolstools) print(final.choices[0].message.content) asyncio.run(main())并行执行时要注意如果某个工具调用失败需要决定是整体回滚还是单独返回错误信息给模型。通常建议把错误信息作为工具结果回传让模型自行判断下一步。7. MCP 协议基础MCPModel Context Protocol是 Anthropic 于 2024 年底开源的标准协议旨在解决“每个应用都要为模型单独适配一套工具接口”的问题。MCP 采用客户端-服务器架构MCP 客户端如 Claude Desktop、IDE 插件连接 MCP 服务器服务器暴露工具、资源和提示词模型通过统一协议调用。MCP 的核心概念包括工具Tools可被模型调用的函数与 Function Calling 中的工具概念一致。资源Resources可被读取的数据如文件内容、数据库记录。提示词Prompts预定义的提示模板帮助模型理解任务。传输层Transports支持 stdio 和 HTTP/SSE 两种通信方式。MCP 使用 JSON-RPC 2.0 作为消息协议所有请求和响应都遵循统一格式。一个典型的 MCP 工具调用流程是客户端发送 tools/call 请求服务器执行并返回结果。8. MCP 实战构建一个最小服务器下面使用官方 Python SDK 构建一个最小 MCP 服务器暴露一个“获取当前时间”的工具。from mcp.server.fastmcp import FastMCP from datetime import datetime mcp FastMCP(TimeServer) mcp.tool() def get_current_time(timezone: str UTC) - str: 获取指定时区的当前时间 # 简化实现实际应使用 zoneinfo 处理时区 return datetime.now().isoformat() if name main: mcp.run(transportstdio)启动后任何支持 MCP 的客户端都可以连接这个服务器并调用 get_current_time 工具。服务器通过装饰器自动生成工具定义SDK 负责处理 JSON-RPC 通信细节。9. MCP 客户端调用实战下面演示如何在 Python 中作为 MCP 客户端连接上述服务器并调用工具。import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def main(): server_params StdioServerParameters( commandpython, args[time_server.py], ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() # 列出可用工具 tools await session.list_tools() print(可用工具:, [t.name for t in tools.tools]) # 调用工具 result await session.call_tool( get_current_time, arguments{timezone: Asia/Shanghai}, ) print(工具结果:, result.content) asyncio.run(main())这个示例展示了 MCP 客户端连接、初始化、列出工具和调用工具的完整流程。实际项目中MCP 客户端通常嵌入在 Agent 框架中由模型根据用户意图自动选择并调用工具。10. 工具调用与 MCP 的对比维度Function CallingMCP定位模型接口层的工具调用能力工具生态的标准化协议工具来源由应用开发者硬编码在请求中由 MCP 服务器动态提供跨应用复用困难每个应用各自适配容易同一服务器可被多客户端复用传输方式HTTP 请求内嵌stdio 或 HTTP/SSE典型场景单应用内快速接入工具多应用共享工具生态、插件市场两者并非互斥MCP 服务器内部暴露的工具最终仍需要通过模型的 Function Calling 能力被调用。可以理解为 MCP 是“工具的分发层”Function Calling 是“模型的调用层”。11. 安全边界工具调用的风险工具调用赋予模型“行动能力”也引入了新的安全风险。主要风险包括提示注入外部内容如网页、邮件中嵌入恶意指令诱导模型调用危险工具。权限滥用模型在用户未授权的情况下调用高权限工具如删除文件、转账。参数篡改模型生成的参数超出预期范围导致数据泄露或系统损坏。过度调用模型在循环中反复调用工具造成资源消耗或费用失控。安全设计应遵循“最小权限”原则每个工具只授予完成任务所需的最小权限并在调用前进行用户确认。12. 安全边界MCP 的防护机制MCP 协议本身提供了一些安全机制但最终安全责任仍在应用层。关键防护点包括工具白名单客户端只暴露必要的工具给模型不暴露全部。用户确认高风险工具删除、写入、支付必须经过用户显式确认。输入校验服务器端对工具参数做严格校验拒绝非法输入。审计日志记录所有工具调用便于事后追溯。沙箱隔离在受限环境中执行工具限制网络和文件系统访问。下面给出一个带用户确认和参数校验的工具调用示例def safe_delete_file(path: str, confirm: bool False) - str: 安全删除文件必须显式确认 if not confirm: return 操作已取消需要用户确认 # 校验路径防止目录穿越 if .. in path or not path.startswith(/data/): return 非法路径 # 实际删除逻辑 return f已删除 {path}这个示例体现了两个关键安全实践高风险操作必须二次确认路径参数必须校验防止目录穿越。13. 实战带安全控制的 Agent下面综合演示一个带安全控制的 Agent模型可以调用工具但高风险工具需要用户确认且所有调用都记录日志。import json import logging from openai import OpenAI logging.basicConfig(levellogging.INFO) logger logging.getLogger(agent) client OpenAI() def send_email(to: str, content: str) - str: 高风险工具发送邮件 # 实际发送逻辑 return f邮件已发送至 {to} def read_file(path: str) - str: 低风险工具读取文件 if .. in path: return 非法路径 return f文件内容: {path} tools [ {type: function, function: {name: send_email, description: 发送邮件, parameters: {type: object, properties: {to: {type: string}, content: {type: string}}, required: [to, content]}}}, {type: function, function: {name: read_file, description: 读取文件, parameters: {type: object, properties: {path: {type: string}}, required: [path]}}}, ] HIGH_RISK_TOOLS {send_email} def execute_tool(name: str, arguments: str) - str: args json.loads(arguments) logger.info(工具调用: %s %s, name, arguments) if name in HIGH_RISK_TOOLS: # 高风险工具需要用户确认 confirm input(f确认执行 {name}? (y/n): ) if confirm.lower() ! y: return 用户取消了操作 if name send_email: return send_email(args[to], args[content]) if name read_file: return read_file(args[path]) return 未知工具 messages [{role: user, content: 读取 config.txt 并发送邮件给 adminexample.com}] for _ in range(5): # 限制最大循环次数防止无限调用 resp client.chat.completions.create(modelgpt-4o, messagesmessages, toolstools) assistant_msg resp.choices[0].message messages.append(assistant_msg) if not assistant_msg.tool_calls: print(最终回答:, assistant_msg.content) break for tc in assistant_msg.tool_calls: result execute_tool(tc.function.name, tc.function.arguments) messages.append({role: tool, tool_call_id: tc.id, content: result})/code/pre 这个示例实现了三个关键安全控制高风险工具的用户确认、工具调用日志审计、最大循环次数限制。这些机制共同构成了工具调用的安全边界。 14. 常见陷阱与最佳实践 在实际开发中工具调用和 MCP 集成有几个常见陷阱需要规避 忘记追加 assistant 消息工具调用后必须把 assistant 消息原样追加回对话否则模型丢失上下文。 tool_call_id 不匹配工具结果必须通过 tool_call_id 与调用请求关联否则模型无法理解结果归属。 参数 Schema 过于宽松缺少 enum、format 约束会导致模型生成非法参数。 无限循环调用必须设置最大迭代次数防止模型反复调用工具。 忽略错误处理工具执行失败时应把错误信息回传给模型而不是直接中断。 最佳实践总结工具定义要精确、参数校验要严格、高风险操作要确认、所有调用要审计、循环要有上限。 15. 总结 工具调用让大模型从“会说话”进化为“能做事”MCP 则让工具生态标准化、可复用。本文从格式、并行和安全三个维度展开给出了完整的代码实战。核心要点是工具定义要结构化、并行调用要处理依赖、安全边界要贯穿始终。建议读者在真实项目中从最小工具集开始逐步扩展并始终把安全控制放在首位。