大模型工具调用与 MCP:格式、并行与安全边界
1. 引言
大模型(LLM)本身是“文本生成器”,无法直接执行外部操作,比如查询数据库、调用 API、读写文件或发送邮件。工具调用(Function Calling / Tool Use)让模型在对话中声明“我需要调用某个工具”,由外部系统真正执行,再把结果回传给模型继续推理。MCP(Model 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(
model="gpt-4o",
messages=messages,
tools=tools,
)
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(
model="gpt-4o",
messages=messages,
tools=tools,
)
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(model="gpt-4o", messages=messages, tools=tools)
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(model="gpt-4o", messages=messages, tools=tools)
print(final.choices[0].message.content)
asyncio.run(main())
并行执行时要注意:如果某个工具调用失败,需要决定是整体回滚还是单独返回错误信息给模型。通常建议把错误信息作为工具结果回传,让模型自行判断下一步。
7. MCP 协议基础
MCP(Model 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(transport="stdio")
启动后,任何支持 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(
command="python",
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 Calling | MCP |
|---|---|---|
| 定位 | 模型接口层的工具调用能力 | 工具生态的标准化协议 |
| 工具来源 | 由应用开发者硬编码在请求中 | 由 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(level=logging.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 并发送邮件给 admin@example.com"}]
for _ in range(5): # 限制最大循环次数,防止无限调用
resp = client.chat.completions.create(model="gpt-4o", messages=messages, tools=tools)
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 则让工具生态标准化、可复用。本文从格式、并行和安全三个维度展开,给出了完整的代码实战。核心要点是:工具定义要结构化、并行调用要处理依赖、安全边界要贯穿始终。建议读者在真实项目中从最小工具集开始,逐步扩展,并始终把安全控制放在首位。
更多推荐

所有评论(0)