简介
到目前为止,我们已经在第一篇博客中理解了为何需要 MCP(Model Context Protocol),并掌握了 Prompt 三角色与 Function Calling(工具调用)的基础操作。接下来,本篇将带你从“用大模型”过渡到“管理语义链”的思维维度,揭开 MCP 的关键数据结构与运行机制。我们会重点讲解:
- MCP 是什么:大模型的语义操作系统MCP 三阶段:Pre / In / Post Model 执行控制MCP 语义角色结构MCP 与 OpenAI、Claude 的 Tool Use 结构对比
通过阅读,你会对 MCP 的核心字段有清晰认识,并且能够用 Python(Pydantic)编写最简版 MCPRequest
、MCPMessage
、MCPPhase
等结构体,为后续实战打下基础。
✅ 1. MCP 是什么:大模型的语义操作系统
核心点:MCP 将用户对话、工具/函数调用、Agent 协作等所有语义交互,视作“大模型的上下文协议层”,从而在调用链条的每一步都可以精确地描述“谁发起、在什么时候、为何而发、如何执行”。
1.1 请求上下文统一封装:who / when / why / how
在没有 MCP 之前,我们的聊天系统上下文往往依赖于纯粹的 messages: [{role, content}]
。但极简的两字段并不能表达诸如“这是哪个 Agent 发起的”、“这是一次工具调用还是普通对话”、“在第几轮”、“携带的元数据”等重要信息。MCP 通过以下几个核心字段进行补充与约定:
actor
(谁发起):指定当前消息或调用的发起者,比如 "user"
、"assistant"
、"calculator_agent"
、"search_agent"
。timestamp
(何时发起):UNIX 时间戳或 ISO 格式,用来记录事件发生时点,方便排查时序问题。phase
(处于哪个阶段):三种取值:"pre_model"
、"in_model"
、"post_model"
,告诉整个系统“当前这条消息要在什么时候生效”。intent
(做什么):类似枚举值,用以区分“纯对话(CHAT)”“工具调用(TOOL_CALL)”“记忆加载(MEMORY_LOAD)”“Agent 路由(AGENT_ROUTE)”等。payload
(实际内容):对于对话场景,就是和原来 messages[i].content
类似;对于函数调用场景,则是函数名 + 参数结构化后的 JSON;对于多 Agent 协作,则可能是整套上下文片段。metadata
(元数据):可以承载任何辅助信息,例如“会话 ID”、“用户等级”、“业务线编号”等,方便后端统一审计、权限控制。MCPRequest 样例(JSON 格式)
{ "actor": "user", "timestamp": 1686200000, "phase": "pre_model", "intent": "CHAT", "payload": { "role": "user", "content": "请帮我推荐一款性价比高的手机。" }, "metadata": { "session_id": "abc123", "user_tier": "premium" }}
这条 MCPRequest 表示:
Who(谁发起): 用户When(何时发起 / 阶段): Pre-Model 阶段(也就是还没把对话内容喂给 LLM 前)Why(意图): 普通聊天(CHAT)How(载荷): 带着
role=user
的消息"请帮我推荐一款性价比高的手机。"
从而,MCP 引擎就能在 Pre-Model 阶段做两件事:
- 验证
metadata.session_id
是否有效、检查权限将 payload
转换成原生的 Chat API messages
列表,发送给 LLM后续一旦模型开始推理(进入 In-Model 阶段),MCP 会依据上下文中的 intent
与 phase
来判断该如何处理函数/工具调用,保证系统提示词不被篡改。
✅ 2. MCP 三阶段:Pre / In / Post Model 执行控制
核心点:MCP 对每次请求都划分为三个阶段,让我们可以在模型推理前、中、后都插入自定义逻辑。这也是 MCP 作为“语义操作系统”最关键的一环。
[ 用户发起 → MCP Pre-Model ] ↓ [ LLM In-Model ] ↓ [ MCP Post-Model ] ↓ [ 反馈给用户/下游 ]
2.1 前置信息注入(Pre-Model)
目标:在调用 LLM 之前,对上下文进行预处理,包括:
- 注入系统提示词(System Prompt)对 Payload 做脱敏、合并、分片等操作加载或检索用户的 Memory(多轮对话记忆、知识库内容等)绑定、拼接必要的元数据(如 A/B-Test 标记、实验 ID)
示例:Pre-Model 阶段的处理流程
from datetime import datetimefrom typing import List, Union, Dictfrom pydantic import BaseModel# 1. 定义 MCPPhase 枚举class MCPPhase(str): PRE_MODEL = "pre_model" IN_MODEL = "in_model" POST_MODEL = "post_model"# 2. 定义 MCPMessage(即 payload 中的 message 结构)class MCPMessage(BaseModel): role: str # "user" / "assistant" / "function" content: str name: str = None # 当 role="function" 时,填函数名# 3. 定义 MCPRequestclass MCPRequest(BaseModel): actor: str timestamp: int phase: MCPPhase intent: str payload: Union[MCPMessage, dict] metadata: Dict[str, Union[str, int, dict]]# 模拟一个用户请求进来raw_request = { "actor": "user", "timestamp": int(datetime.now().timestamp()), "phase": MCPPhase.PRE_MODEL, "intent": "CHAT", "payload": {"role": "user", "content": "帮我写一段 Python 代码,计算斐波那契数列前 10 项。"}, "metadata": {"session_id": "session_001", "user_level": "free"}}# 预处理逻辑示例def mcp_preprocess(request: MCPRequest) -> List[dict]: """ Pre-Model 阶段逻辑: 1. 验证 session_id 2. 注入系统提示词 3. 加载 Memory(这里先假设没有历史) 4. 拼装成 OpenAI 的 messages 列表 """ # 验证 session_id(示例里直接通过) assert request.metadata.get("session_id"), "缺少 session_id" # 注入系统提示词 system_msg = MCPMessage(role="system", content="你是一名专业的 Python 编程助手。") # 构造最终要发送给 LLM 的 messages messages = [system_msg.dict(), request.payload] # payload 本身就是一个 MCPMessage return messages# 演示调用parsed_request = MCPRequest.parse_obj(raw_request)llm_messages = mcp_preprocess(parsed_request)print("Pre-Model 拼装的 messages:", llm_messages)
运行结果示例(打印):
[ {"role": "system", "content": "你是一名专业的 Python 编程助手。"}, {"role": "user", "content": "帮我写一段 Python 代码,计算斐波那契数列前 10 项。"}]
这段代码演示了如何在 Pre-Model 阶段,将最初的 MCPRequest.payload
(用户消息)与系统提示词合并,变成原生的 Chat API messages
。在这个阶段,你还可以加入“检索 Memory”、“检查黑白名单”、“做内容安全校验”等逻辑,并且只要 phase = PRE_MODEL
,就说明这些逻辑不会影响到 LLM 推理中对“模型运行状态”的控制。
2.2 模型运行中响应控制(In-Model)
目标:当 LLM 在推理过程中,如果遇到需要调用工具/函数、需要中断推理或注入额外信息之类的指令,就会进入 In-Model 阶段。在这一阶段,MCP 主要负责:
- 解析 LLM 响应里是否带有
function_call
或自定义 “TOOL_CALL” 标记判断是否需要将请求转为下一轮的 MCPRequest(并且 phase 依旧是 In-Model),例如对工具调用过程做递归嵌套对于多 Agent 场景,In-Model 期间还会进行“上下文路由”——把部分上下文发给其他 Agent示例:In-Model 处理 Function Calling
继承上节示例,假设我们在 Pre-Model 阶段拼装好了 messages
并发送给 LLM,得到了下面的响应:
{ "choices": [ { "message": { "role": "assistant", "content": null, "function_call": { "name": "calculate_fib", "arguments": "{"n": 10}" } } } ]}
In-Model 阶段拆解流程
MCP 解析到 LLM 返回了带 function_call
的消息,说明模型“要调用工具”。
MCP 根据 function_call.name
和 arguments
构造一个新的 MCPRequest:
actor = "assistant"
phase = MCPPhase.IN_MODEL
intent = "TOOL_CALL"
payload = {"name": "calculate_fib", "arguments": {"n":10}}
Python 端执行 calculate_fib(n=10)
得到列表 [0,1,1,2,3,5,8,13,21,34]
。
MCP 再构造一条 role="function"
的 MCPMessage,把结果放在 payload
里,并把 phase
设为 "in_model"
,继续下一轮调用。
# 继续在前面示例的基础上,演示In-Model的伪代码# 1. 模拟 LLM 返回(带 function_call)llm_response = { "choices": [ { "message": { "role": "assistant", "content": None, "function_call": { "name": "calculate_fib", "arguments": "{"n": 10}" } } } ]}def mcp_handle_in_model(request: MCPRequest, llm_message: dict): """ 处理 LLM 输出的 function_call, 1. 构造 Tool-Call MCPRequest 2. 执行工具 3. 构造新的 In-Model 阶段的 function 返回消息 """ func_call = llm_message["function_call"] func_name = func_call["name"] func_args = json.loads(func_call["arguments"]) # 2. 执行用户定义的函数(这里用伪函数演示) if func_name == "calculate_fib": n = func_args["n"] # 计算前n项斐波那契 fib = [0, 1] for i in range(2, n): fib.append(fib[-1] + fib[-2]) result = fib[:n] else: result = None # 3. 构造一条新的 MCPMessage,用作 LLM 下一轮的输入 function_message = MCPMessage( role="function", name=func_name, content=json.dumps({"fib_sequence": result}) ) # 注意:phase 依然保持 In-Model,因为工具调用仍在“推理链”中 next_request = MCPRequest( actor="assistant", timestamp=int(datetime.now().timestamp()), phase=MCPPhase.IN_MODEL, intent="TOOL_CALL", payload=function_message.dict(), metadata=request.metadata # 原始 metadata 可延续 ) return next_request# 演示调用parsed_pre_request = parsed_request # 来自上节 Pre-Model 的 parsed_request# 1. LLM 给出的 llm_messagellm_message = llm_response["choices"][0]["message"]# 2. MCP 处理 In-Modelin_model_request = mcp_handle_in_model(parsed_pre_request, llm_message)print("In-Model 阶段构造的下一轮请求:", in_model_request.json(indent=2, ensure_ascii=False))
输出示例(In-Model 构造的下一轮 MCPRequest):
{ "actor": "assistant", "timestamp": 1686201234, "phase": "in_model", "intent": "TOOL_CALL", "payload": { "role": "function", "content": "{"fib_sequence": [0,1,1,2,3,5,8,13,21,34]}", "name": "calculate_fib" }, "metadata": { "session_id": "session_001", "user_level": "free" }}
此时,MCP 就将“函数调用结果”以 role="function"
的形式包装在 payload
中。如果后续 LLM 还需要读入这份斐波那契序列,就会继续下一次 In-Model 调用。直到没有 function_call
,或 intent
变更,才会进入 Post-Model 阶段。
2.3 模型输出后的后处理(Post-Model)
目标:当 LLM 不再主动发出工具/函数调用时,说明已经可以产出最终自然语言回答。这时进入 Post-Model 阶段,MCP 负责做:
- 对 LLM 最终输出进行业务路由(例如日志落盘、用户数据更新)如果
intent
是“需要额外持久化 Memory”,就在此阶段更新数据库如果 intent
是“要发起下游系统调用”(比如发订单、发邮件),则在此处触发实际的 API 请求最后把 LLM 输出或工具链组合结果,统一转换为对前端可读的格式,交给调用方示例:Post-Model 阶段的业务处理
假设 LLM 最终回答已经出来,MCP 收到:
{ "choices": [ { "message": { "role": "assistant", "content": "前 10 项斐波那契数列为 [0,1,1,2,3,5,8,13,21,34]。" } } ]}
Post-Model 整体流程
- MCP 读取这条没有
function_call
的消息,发现 intent="CHAT"
,表明不需要继续工具调用。检查 Metadata:如果 metadata
里有 need_persist_memory = true
,则把本次对话内容写入 Redis 或数据库。打日志:记录 “session_id=xxx,本轮对话正常结束”拼装对前端的输出:把原生 LLM message.content
拿出来,构造成约定格式(比如仅返回 {"response": "..."}"
)。# 伪代码示例def mcp_postprocess(request: MCPRequest, llm_message: dict): """ 处理 Post-Model 阶段:持久化、路由、日志、输出封装 """ # 1. 如果需要写 Memory if request.intent == "CHAT" and request.metadata.get("need_persist_memory"): save_to_memory(request.metadata["session_id"], llm_message["content"]) # 2. 打日志 print(f"[MCP POST] 会话 {request.metadata['session_id']} 正常结束,输出:{llm_message['content']}") # 3. 构造给前端的标准输出格式 return {"response": llm_message["content"]}# 演示调用final_llm_message = {"role": "assistant", "content": "前 10 项斐波那契数列为 [0,1,1,2,3,5,8,13,21,34]。"}post_output = mcp_postprocess(in_model_request, final_llm_message)print("最终返回给前端:", post_output)
输出示例:
[MCP POST] 会话 session_001 正常结束,输出:前 10 项斐波那契数列为 [0,1,1,2,3,5,8,13,21,34]。最终返回给前端: {'response': '前 10 项斐波那契数列为 [0,1,1,2,3,5,8,13,21,34]。'}
综上,MCP 的“三阶段”将一个原本“黑箱式调用 LLM → 得到回答”的流程,细分成 Pre-Model 注入逻辑、In-Model 工具/函数调用控制、Post-Model 结果处理与路由。这样一来,无论是在性能监控、日志审计、还是安全脱敏、跨 Agent 协作时,都可以精准控制在哪个阶段干什么。
✅ 3. MCP 语义角色结构
核心点:在 MCP 中,除了最基本的
role: user/assistant/function
,我们要进一步给每条消息加上actor
、intent
、tool_use
等属性,形成“多维度”的语义描述。
3.1 主要字段说明
下面是 MCPMessage(作为 payload
的载体)中常见的部分字段及含义示例:
字段 | 含义 |
---|---|
actor | 执行实体:如 "user" 、"assistant" (LLM 本身)、"search_agent" 、"calculator_agent" 等 |
intent | 意图类型:如 "CHAT" 、"TOOL_CALL" 、"MEMORY_LOAD" 、"AGENT_ROUTE" 等 |
tool_use | 是否是一次工具调用:true 或 false ;或者用更细粒度的枚举 { "name": "search" } 等 |
message | 嵌套的对话消息:包含 role: user/assistant/function 、content 、name (函数名)等 |
metadata | 上下文元信息:如 session_id 、conversation_id 、priority 、A/B_test_tag 、user_tier 等 |
示例:一个包含多维度语义的 MCPMessage
{ "actor": "assistant", "intent": "TOOL_CALL", "tool_use": {"name": "product_search"}, "message": { "role": "assistant", "name": null, "content": "正在帮您搜索性价比高的手机..." }, "metadata": { "session_id": "session_001", "region": "CN", "user_tier": "free" }}
- actor:说明这条消息是哪个执行实体发出的。intent:告诉处理逻辑,这是一次“工具调用”类型的交互。tool_use:更明确地指出要调用的工具名称
product_search
。message:包含了原本「assistant 角色要说的话」,在这里是 "正在帮您搜索性价比高的手机..."
。metadata:额外记录区域、用户等级等,不在 LLM 上下文里直接展示,但对业务逻辑很关键。通过这种多维度结构,我们就可以在 MCP Pre/In/Post 三个阶段灵活地做过滤、审计、路由、脱敏等操作。例如,在 Pre-Model 阶段,如果 intent == "TOOL_CALL"
且 tool_use.name == "payment_agent"
,我们就必须先校验用户余额;如果 intent == "CHAT"
且 metadata.user_tier == "free"
,就需要给 LLM 加一个“免费用户每分钟最多 10 次调用”的系统提示。
✅ 4. MCP 与 OpenAI、Claude 的 Tool Use 结构对比
核心点:市面上常见的 OpenAI Function Calling、Claude Tool Use v1/v2,都只是 MCP 思想的“特定实现”,它们在 JSON 结构、管控节点上存在差异。理解这些差异,有助于我们在自己实现 MCP 时兼容不同服务。
4.1 OpenAI Function Calling 的 JSON 结构
以 OpenAI 为例,我们在调用 ChatCompletion.create()
时,functions
、function_call
字段就是 Function Calling 的核心。其典型的输入/输出形式如下:
请求端(用户 → LLM)
{ "model": "gpt-4o-mini", "messages": [ { "role": "system", "content": "你是一名 AI 助手。" }, { "role": "user", "content": "请计算 6 * 7 的结果。" } ], "functions": [ { "name": "multiply", "description": "将两个整数相乘", "parameters": { "type": "object", "properties": { "a": { "type": "integer" }, "b": { "type": "integer" } }, "required": ["a", "b"] } } ], "function_call": "auto"}
模型返回(LLM → 用户)
{ "choices": [ { "message": { "role": "assistant", "content": null, "function_call": { "name": "multiply", "arguments": "{"a":6,"b":7}" } } } ]}
函数执行结果回传(用户 → LLM)
{ "model": "gpt-4o-mini", "messages": [ { "role": "system", "content": "你是一名 AI 助手。" }, { "role": "user", "content": "请计算 6 * 7 的结果。" }, { "role": "assistant", "content": null, "function_call": { "name": "multiply", "arguments": "{"a":6,"b":7}" } }, { "role": "function", "name": "multiply", "content": "{"result": 42}" } ]}
以上结构在实现 MCP 所需的核心要素时,还缺少以下几点:
- actor 字段:区分到底是哪个 Agent 在发起——Function Calling 只能以
role="assistant"
表示模型本身发起。phase 阶段:在 JSON 里并没有明确标注“Pre/In/Post”哪个阶段。更多元的 intent 与 metadata:OpenAI 语法里只能靠拼 role + function_call
来隐式表达意图,并无法携带额外的 metadata。4.2 Claude Tool Use v1 / v2 的 JSON 结构
以 Anthropic Claude 为例,其 Tool Use v1、v2 在设计上更接近 MCP 思想,但依然不是完全通用的协议。
V1 结构(示例)
{ "options": { "model": "claude-2-opus", "plugins": [ { "plugin_id": "my_search_tool", "arguments": { "query": "MCP 是什么?" } } ] }, "messages": [ { "speaker": "user", "text": "请告诉我 MCP 的核心机制。" } ]}
- 优点:在
plugins
数组里直接列出“要调用的工具”和“参数”缺点:这种结构只能支持“一次性调用一个或多个”,并不明确标注“阶段”、也没有 actor
、intent
等字段。V2 结构(示例)
{ "messages": [ { "speaker": "system", "text": "你是一个 API 协调专家。" }, { "speaker": "assistant", "text": "{{TOOL_CALL: search({"query": "MCP 语义协议"})}}" } ]}
- 优点:在
text
中出现特定占位符,就能自动触发相应工具调用。缺点:这种占位符方式依赖“正则匹配”或“字符串解析”,并不是真正的 JSON 字段,缺乏结构化的定义,且容易出错。4.3 用 Python 比较三者差异
下面给出一个示例脚本,将 OpenAI Function Calling、Claude Tool Use v1、MCP 三种方式并列展示,帮助你直观感受各自的结构特点。
from pydantic import BaseModelfrom typing import List, Dict, Any# 1. OpenAI Function Calling 示例openai_example = { "model": "gpt-4o-mini", "messages": [ {"role": "system", "content": "你是一名 AI 助手。"}, {"role": "user", "content": "请计算 6 * 7 的结果。"} ], "functions": [ { "name": "multiply", "description": "将两个整数相乘", "parameters": { "type": "object", "properties": { "a": {"type": "integer"}, "b": {"type": "integer"} }, "required": ["a", "b"] } } ], "function_call": "auto"}# 2. Claude Tool Use v1 示例claude_v1_example = { "options": { "model": "claude-2-opus", "plugins": [ { "plugin_id": "my_search_tool", "arguments": {"query": "MCP 是什么?"} } ] }, "messages": [ {"speaker": "user", "text": "请告诉我 MCP 的核心机制。"} ]}# 3. MCP 示例class MCPMessage(BaseModel): actor: str timestamp: int phase: str intent: str payload: Any metadata: Dict[str, Any]mcp_example = MCPMessage( actor="user", timestamp=1686205000, phase="pre_model", intent="CHAT", payload={"role": "user", "content": "请计算 6 * 7 的结果。"}, metadata={"session_id": "abc123", "user_tier": "free"}).dict()# 并列打印import jsonprint("=== OpenAI Function Calling 示例 ===")print(json.dumps(openai_example, ensure_ascii=False, indent=2))print("\n=== Claude Tool Use v1 示例 ===")print(json.dumps(claude_v1_example, ensure_ascii=False, indent=2))print("\n=== MCP 示例 ===")print(json.dumps(mcp_example, ensure_ascii=False, indent=2))
运行结果示例:
=== OpenAI Function Calling 示例 ==={ "model": "gpt-4o-mini", "messages": [ { "role": "system", "content": "你是一名 AI 助手。" }, { "role": "user", "content": "请计算 6 * 7 的结果。" } ], "functions": [ { "name": "multiply", "description": "将两个整数相乘", "parameters": { "type": "object", "properties": { "a": { "type": "integer" }, "b": { "type": "integer" } }, "required": ["a", "b"] } } ], "function_call": "auto"}=== Claude Tool Use v1 示例 ==={ "options": { "model": "claude-2-opus", "plugins": [ { "plugin_id": "my_search_tool", "arguments": { "query": "MCP 是什么?" } } ] }, "messages": [ { "speaker": "user", "text": "请告诉我 MCP 的核心机制。" } ]}=== MCP 示例 ==={ "actor": "user", "timestamp": 1686205000, "phase": "pre_model", "intent": "CHAT", "payload": { "role": "user", "content": "请计算 6 * 7 的结果。" }, "metadata": { "session_id": "abc123", "user_tier": "free" }}
从并列示例可以看到:
- OpenAI Function Calling:结构里只有
messages/functions/function_call
,缺失明确的 actor
、phase
、intent
、metadata
等字段。Claude Tool Use v1:把工具作为 “plugins” 放在 options
中,但并没有在对话上下文里标记“使用工具”的时机,也没有细粒度的元数据。MCP:一行 JSON 就将“谁发起、何时在哪个阶段、意图为何、载荷是什么、额外属性有哪些”全部表达清楚,更具可扩展性与可控性。✅ 阶段产出
了解 MCP 的核心字段结构(消息上下文 + 调用上下文 + 控制上下文)
- 你已经清楚地掌握了 MCP 中
actor
、timestamp
、phase
、intent
、payload
、metadata
等字段的意义,以及它们如何在 Pre/In/Post 三个阶段协同工作。在每个阶段,你都可以依赖这些字段来做“权限检查”、“工具路由”、“记忆加载/写入”、“日志审计”等操作。自己写一个 Python MCPRequest 数据结构(推荐用 Pydantic)
- 本文示例里,我们已经用 Pydantic 定义了最简版的
MCPPhase
、MCPMessage
、MCPRequest
。你可以在此基础上,根据自己的业务扩展更多字段,比如 priority
、trace_id
、parent_request_id
、agent_chain
等,从而让 MCP 请求更贴合实际场景需求。至此,通过第②篇博客的学习,你已经从“单纯用大模型”升级到“管理语义链”的思维模式,为后续“实战搭建 MCP 层”和“构建多 Agent 协作”打下了坚实基础。下一篇我们将进入 第三阶段:用 Python 实现一个最小 MCP 系统,带你动手把 MCP 的数据结构、调用转换、工具链调用等整合到一起,让模型真正“像一个有操作系统的 Agent”来工作。敬请期待!