大模型工具调用能力之二——MCP 模型上下文协议 :AI 生态的标准化连接协议,打通模型与外部系统
在AI智能体开发中,Function Calling(FC) 是大模型调用外部工具的事实标准,但手写FC的JSON格式繁琐、易出错、维护成本高。
FastMCP 作为MCP协议的Python极简实现,底层核心就是封装标准化Function Calling,通过极简装饰器语法,全自动完成FC的JSON构造、解析、响应全流程,让开发者无需关心底层协议细节,快速搭建可被LLM调用的工具服务。
一、MCP核心本质:底层基于Function Calling
MCP并非全新的调用机制,其工具调用的底层通信逻辑完全依托标准Function Calling实现。
简单来说:
- Function Calling:LLM与外部工具交互的底层标准协议,基于固定JSON格式完成指令传递;
- MCP:在FC基础上扩展的上下文管理、多工具协同上层规范;
最常用的实现框架,FastMCP:
将MCP规范+FC的JSON格式全自动化封装,用Python装饰器替代手写JSON,零门槛实现工具服务。
二、FastMCP架构(核心的客户端和服务端,不介绍host)
- Client(客户端):调用方,即支持Function Calling的LLM(通义千问/GPT/Claude等)、本地测试程序;
- FastMCP Server(服务端):框架核心,自动完成FC-JSON封装、参数校验、工具执行、响应格式化。
三、核心重点:FastMCP ↔ Function Calling ↔ JSON 映射关系
FastMCP的核心原理,先明确原生Function Calling的JSON格式,再对应FastMCP的封装逻辑,理清三者关系。
3.1 原生Function Calling标准JSON格式
Function Calling的交互分为工具定义和调用请求两个核心JSON结构,这是LLM识别的底层语法:
(1)工具定义JSON(告诉LLM有哪些工具可用)
[
{
"type": "function",
"function": {
"name": "calculator",
"description": "计算器工具,支持加减乘除",
"parameters": {
"type": "object",
"properties": {
"num1": {"type": "number", "description": "第一个数字"},
"num2": {"type": "number", "description": "第二个数字"},
"operator": {"type": "string", "description": "运算符+-*/"}
},
"required": ["num1", "num2", "operator"]
}
}
}
]
(2)调用请求JSON(LLM让程序执行工具)
{
"role": "assistant",
"content": null,
"tool_calls": [
{
"function": {
"name": "calculator",
"arguments": "{\"num1\":10,\"num2\":6.5,\"operator\":\"*\"}"
}
}
]
}
3.2 FastMCP与FC-JSON的自动映射(核心)
FastMCP通过@mcp.tool()装饰器,将Python函数自动转换为上述FC标准JSON,全程无需手写JSON,映射关系如下:
|
Python代码元素 |
FastMCP封装后 |
对应Function Calling JSON字段 |
|
函数名 |
工具唯一标识 |
|
|
函数文档字符串 |
工具功能描述 |
|
|
参数类型注解 |
参数数据类型 |
|
|
参数名 |
参数键名 |
|
|
必传参数 |
必填约束 |
|
|
函数返回值 |
工具执行结果 |
LLM整合的FC响应内容 |
一句话总结:
你写的Python函数 → FastMCP自动转成FC标准JSON → LLM解析JSON并下发调用指令 → FastMCP解析指令执行函数 → 返回结果给LLM。
四、FastMCP传输方式:stdio 与 SSE
FastMCP基于Function Calling的JSON指令,提供两种传输方式,用于传递FC格式数据,适配不同场景:
4.1 stdio(标准输入输出)
- 原理:基于本地进程的标准输入/输出流,直接传输FC-JSON数据;
- 特点:无网络开销、轻量高效、仅本地通信;
- 适用场景:LLM与FastMCP服务在同一台机器/同一进程,本地开发、测试首选;
- 底层:通过stdin/stdout传递FC格式的JSON字符串。
4.2 SSE(Server-Sent Events)
- 原理:基于HTTP长连接的服务器推送,将FC-JSON数据通过网络流式传输;
- 特点:跨网络、支持远程调用、兼容云端LLM;
- 适用场景:LLM部署在云端,FastMCP服务在本地/服务器,生产环境首选;
- 底层:通过HTTP接口传输FC格式的JSON数据,实现远程工具调用。
五、FastMCP与LLM的完整交互流程
基于Function Calling底层协议,FastMCP与LLM的交互全流程如下,每一步都围绕FC-JSON展开:
- 启动FastMCP服务
服务端扫描所有@mcp.tool()装饰的函数,自动生成FC工具定义JSON,等待LLM连接。 - LLM连接并获取工具列表
LLM通过stdio/SSE连接FastMCP,获取FC格式的工具定义JSON,知晓可用工具。 - 用户发起业务请求
例:10乘以6.5等于多少? - LLM生成FC调用JSON
LLM解析用户意图,生成FC调用请求JSON,通过传输方式发送给FastMCP。 - FastMCP解析并执行工具
服务端解析FC-JSON中的工具名、参数,自动调用对应的Python函数,执行业务逻辑。 - 返回FC响应结果
FastMCP将函数执行结果封装为FC标准响应格式,回传给LLM。 - LLM生成最终回答
LLM结合FC工具返回结果,整理为自然语言,回复给用户。
六、完整可运行实战代码
6.1 FastMCP 三大核心 API(装饰器)
FastMCP 封装了 3 个开发必备的装饰器,全自动对接 MCP/FC 协议:
- @mcp.tool() → 工具注册(底层对应 Function Calling)
用于注册计算器、查天气等业务工具,LLM 核心调用对象。 - @mcp.prompt() → 动态提示词模板
用于定义系统角色、交互规则,约束 LLM 行为。 - @mcp.resource() → 静态资源 / 知识库
注册固定文本、知识库、配置信息,供 LLM 读取使用。
6.2 代码实现(服务端)
# server.py
from fastmcp import FastMCP
from datetime import datetime
import socket
def demo_log(message: str) -> None:
"""演示日志:统一时间戳,便于投屏讲解调用流程。"""
now = datetime.now().strftime("%H:%M:%S")
print(f"[MCP-DEMO {now}] {message}")
def pick_available_port(start_port: int = 8000, max_tries: int = 20) -> int:
"""从 start_port 开始找可用端口,演示时避免端口冲突中断。"""
for port in range(start_port, start_port + max_tries):
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
if sock.connect_ex(("127.0.0.1", port)) != 0:
return port
raise RuntimeError(f"未找到可用端口(尝试范围: {start_port}-{start_port + max_tries - 1})")
# ===================== 1. 创建FastMCP服务端 =====================
mcp_server = FastMCP(
name="FastMCP完整服务",
version="1.0.0",
instructions="包含tool/prompt/resource,支持客户端工具发现"
)
# ===================== 2. 注册工具(@tool) =====================
@mcp_server.tool()
def calculator(num1: float, num2: float, operator: str) -> str:
"""
计算器工具,支持加减乘除
:param num1: 第一个数字
:param num2: 第二个数字
:param operator: 运算符 + - * /
"""
demo_log(f"tool.calculator 收到请求: num1={num1}, num2={num2}, operator='{operator}'")
try:
if operator == "+":
result = f"计算结果:{num1 + num2}"
elif operator == "-":
result = f"计算结果:{num1 - num2}"
elif operator == "*":
result = f"计算结果:{num1 * num2}"
elif operator == "/":
result = f"计算结果:{num1 / num2}"
else:
result = "错误:仅支持 + - * /"
demo_log(f"tool.calculator 返回结果: {result}")
return result
except Exception as e:
error_msg = f"工具执行失败:{str(e)}"
demo_log(f"tool.calculator 发生异常: {error_msg}")
return error_msg
@mcp_server.tool()
def weather_query(city: str) -> str:
"""查询指定城市天气"""
demo_log(f"tool.weather_query 收到请求: city='{city}'")
weather_data = {
"北京": "晴,25℃,微风",
"上海": "多云,28℃,湿度65%",
"广州": "小雨,26℃"
}
result = weather_data.get(city, "暂无该城市天气")
demo_log(f"tool.weather_query 返回结果: {result}")
return result
# ===================== 3. 注册提示词(@prompt) =====================
@mcp_server.prompt()
def system_prompt(user: str = "用户") -> str:
"""系统提示词模板"""
prompt_text = f"你是{user}的智能助手,必须使用工具回答,禁止编造信息"
demo_log(f"prompt.system_prompt 被调用: user='{user}'")
return prompt_text
# ===================== 4. 注册静态资源(@resource) =====================
@mcp_server.resource("info://bot")
def bot_info() -> str:
"""机器人基础信息"""
data = "我是基于FastMCP开发的智能助手,支持工具调用与知识库查询"
demo_log("resource.bot_info 被读取")
return data
# ===================== 5. 启动服务(SSE模式,支持远程客户端连接) =====================
if __name__ == "__main__":
demo_log("服务初始化完成")
demo_log("已注册: tools=[calculator, weather_query]")
demo_log("已注册: prompt=[system_prompt]")
demo_log("已注册: resource=[info://bot]")
port = pick_available_port(start_port=8000, max_tries=30)
if port != 8000:
demo_log(f"检测到 8000 端口被占用,自动切换到 {port}")
demo_log(f"启动 transport=sse,监听 http://0.0.0.0:{port}/sse")
mcp_server.run(transport="sse", host="0.0.0.0", port=port)
[MCP-DEMO 17:37:01] 服务初始化完成
[MCP-DEMO 17:37:01] 已注册: tools=[calculator, weather_query]
[MCP-DEMO 17:37:01] 已注册: prompt=[system_prompt]
[MCP-DEMO 17:37:01] 已注册: resource=[info://bot]
[MCP-DEMO 17:37:03] 检测到 8000 端口被占用,自动切换到 8002
[MCP-DEMO 17:37:03] 启动 transport=sse,监听 http://0.0.0.0:8002/sse
[04/06/26 17:37:03] INFO Starting MCP server 'FastMCP完整服务' with transport 'sse' on server.py:1358
http://0.0.0.0:8002/sse/
D:\Python\Python312\Lib\site-packages\websockets\legacy\__init__.py:6: DeprecationWarning: websockets.legacy is deprecated; see https://websockets.readthedocs.io/en/stable/howto/upgrade.html for upgrade instructions
warnings.warn( # deprecated in 14.0 - 2024-11-09
D:\Python\Python312\Lib\site-packages\uvicorn\protocols\websockets\websockets_impl.py:17: DeprecationWarning: websockets.server.WebSocketServerProtocol is deprecated
from websockets.server import WebSocketServerProtocol
INFO: Started server process [27296]
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: Uvicorn running on http://0.0.0.0:8002 (Press CTRL+C to quit)
6.3 代码实现(客户端)
客户端通过 list_tools() 方法自动获取服务端所有工具,模拟 LLM 的工具发现逻辑
import asyncio
from datetime import datetime
from fastmcp.client import Client
def demo_log(message: str) -> None:
now = datetime.now().strftime("%H:%M:%S")
print(f"[MCP-CLIENT {now}] {message}")
# ===================== 1. 创建FastMCP客户端(SSE) =====================
async def create_client(start_port: int = 8000, max_tries: int = 30):
"""
自动探测可连接的 SSE MCP 服务端(与 MCP-server.py 的端口策略匹配)。
"""
last_error = None
for port in range(start_port, start_port + max_tries):
url = f"http://127.0.0.1:{port}/sse"
client = Client(url, timeout=8, init_timeout=8)
try:
await client._connect()
demo_log(f"已连接服务端: {url}")
return client, url
except Exception as e:
last_error = e
await client.close()
raise RuntimeError(f"SSE 服务端连接失败,尝试端口 {start_port}-{start_port + max_tries - 1}") from last_error
# ===================== 2. 核心:客户端自动发现服务端工具 =====================
async def discover_tools(client):
"""
工具自动发现:客户端获取服务端所有注册的工具
对应MCP协议:tools/list 方法
底层:获取服务端生成的FC-JSON工具列表
"""
print("=" * 60)
demo_log("客户端开始自动发现服务端工具...")
tools = await client.list_tools() # 核心方法:工具发现
# 打印所有工具信息
for tool in tools:
print(f"\n工具名称:{tool.name}")
print(f"工具描述:{tool.description}")
print(f"输入参数:{tool.inputSchema}")
print("=" * 60)
return tools
# ===================== 3. 客户端调用服务端工具 =====================
async def call_server_tools(client):
print("\n客户端调用服务端工具:")
# 调用计算器工具
calc_result = await client.call_tool(
"calculator",
{"num1": 10, "num2": 6.5, "operator": "*"}
)
print(f"1) 计算器结果:{calc_result}")
# 调用天气查询工具
weather_result = await client.call_tool(
"weather_query",
{"city": "北京"}
)
print(f"2) 天气查询结果:{weather_result}")
# ===================== 4. 客户端调用提示词/资源 =====================
async def call_prompt_and_resource(client):
print("\n客户端调用提示词与资源:")
prompt = await client.get_prompt("system_prompt", {"user": "小明"})
print(f"1) 系统提示词:{prompt.messages}")
resource = await client.read_resource("info://bot")
print(f"2) 静态资源:{resource}")
# ===================== 主流程 =====================
async def main():
client, url = await create_client()
try:
demo_log(f"当前连接地址: {url}")
# 1. 自动发现工具(核心)
await discover_tools(client)
# 2. 调用工具
await call_server_tools(client)
# 3. 调用提示词/资源
await call_prompt_and_resource(client)
finally:
await client.close()
demo_log("客户端已断开")
if __name__ == "__main__":
asyncio.run(main())
[MCP-CLIENT 17:40:59] 已连接服务端: http://127.0.0.1:8000/sse
[MCP-CLIENT 17:40:59] 当前连接地址: http://127.0.0.1:8000/sse
============================================================
[MCP-CLIENT 17:40:59] 客户端开始自动发现服务端工具...
工具名称:calculator
工具描述:计算器工具,支持加减乘除
:param num1: 第一个数字
:param num2: 第二个数字
:param operator: 运算符 + - * /
输入参数:{'properties': {'num1': {'title': 'Num1', 'type': 'number'}, 'num2': {'title': 'Num2', 'type': 'number'}, 'operator': {'title': 'Operator', 'type': 'string'}}, 'required': ['num1', 'num2', 'operator'], 'type': 'object'}
工具名称:weather_query
工具描述:查询指定城市天气
输入参数:{'properties': {'city': {'title': 'City', 'type': 'string'}}, 'required': ['city'], 'type': 'object'}
============================================================
客户端调用服务端工具:
1) 计算器结果:[TextContent(type='text', text='计算结果:65.0', annotations=None)]
2) 天气查询结果:[TextContent(type='text', text='晴,25℃,微风', annotations=None)]
客户端调用提示词与资源:
1) 系统提示词:[PromptMessage(role='user', content=TextContent(type='text', text='你是小明的智能助手,必须使用工具回答,禁止编造信息', annotations=None))]
2) 静态资源:[TextResourceContents(uri=AnyUrl('info://bot'), mimeType='text/plain', text='我是基于FastMCP开发的智能助手,支持工具调用与知识库查询')]
[MCP-CLIENT 17:40:59] 客户端已断开
七、工程化对接LLM
- 启动服务
-
- 本地调用:
mcp.run()(stdio模式,传输FC-JSON) - 远程调用:
mcp.run(transport="sse", port=8000)(SSE模式)
- 本地调用:
- LLM配置
所有支持Function Calling的LLM(通义千问、GPT-4o、Claude等),直接连接FastMCP服务地址,即可自动识别FC工具列表,完成工具调用。
八、总结
- FastMCP底层核心:基于标准Function Calling,所有交互都是FC格式的JSON;
- JSON映射关系:FastMCP通过装饰器自动将Python函数转为FC工具定义JSON,无需手写;
- 传输方式:stdio(本地高效)、SSE(远程生产),均用于传递FC-JSON数据;
- LLM交互:围绕FC协议完成「工具发现→指令下发→执行响应」全流程;
- 开发优势:彻底告别繁琐的FC-JSON手写,几行代码搭建兼容所有LLM的工具服务。
更多推荐


所有评论(0)