MCP 内部机制：JSON-RPC 消息、FastMCP 服务器与 ReAct 追踪机制
通过FastMCP服务器、LangGraph ReAct智能体以及记录每一步操作的跟踪器，遵循工具/列表中的模型上下文协议请求，最终生成答案。
模型上下文协议，即MCP，是Anthropic推出的一项开放标准，它使得Claude等AI模型能够访问外部工具、数据源和服务。一个便于理解的类比是通用插头：无需为每个工具编写定制的集成代码，模型可通过标准方式与外部系统进行交互。
组成结构
该架构由三部分组成。MCP主机是用户实际使用的应用程序，无论是聊天窗口还是集成开发环境，其中包含负责管理连接的客户端。MCP协议则是主机与服务器之间的标准化通信层。每台MCP服务器会连接某一类资源，例如数据库、外部API或代码执行环境，并充当连接它们的桥梁。
运行时的工作流程
从宏观层面来看，请求会经历以下几个阶段：
- 用户向主机发送查询请求。
- 主机询问MCP服务器它们提供了哪些工具。
- 服务器返回各自的工具列表。
- 主机将查询请求及工具列表一起发送给大语言模型。
- 大语言模型选定一个工具并告知主机。
- 主机调用对应的目标MCP服务器来执行该工具。
- 服务器执行工具并返回结果。
- 主机将该结果传回给大语言模型。
- 大语言模型生成最终答案，由主机将其输出。
服务器会提供三种类型的资源。工具是模型可以调用的功能，例如search_web、query_database或send_email。资源则是主机可以获取作为上下文的可读数据，比如文件、数据库记录、日历事件和文档，模型可以读取这些数据但无法执行相关操作。提示则是服务器提供的预写模板，例如GitHub服务器可能会提供一个“总结此PR”的模板，供主机展示给用户。
归根结底，MCP只是主机与服务器进行交互的一种标准化方式。它仅规定了消息格式和对话规则，除此之外并无其他内容。具体而言，它定义了四项内容。
- 消息格式。所有内容均遵循 JSON-RPC 2.0 标准。请求包含
id、method和params，而响应则会重复id，并附上result或error，这样双方就能以统一的方式交流。 - 可用方法。该协议规定了主机可以使用的固定调用集合，例如用于发现功能的
tools/list、用于运行工具的tools/call、用于获取数据的resources/read，以及用于获取模板的prompts/get。服务器必须实现这些方法。 - 握手流程。在首次连接时，双方会交换
initialize消息，声明各自的协议版本及功能，这样在任何操作开始之前，双方就能知晓对方支持的功能。
一次工具调用会拆分为四条消息。首先，主机会询问有哪些工具可用：
{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/list",
"params": {}
}
服务器会回复其工具的描述：
{
"jsonrpc": "2.0",
"id": 1,
"result": {
"tools": [
{
"name": "search_web",
"description": "Search the web for a given query",
"inputSchema": {
"type": "object",
"properties": {
"query": { "type": "string" }
},
"required": ["query"]
}
}
]
}
}
主持人随后叫住了其中一人：
{
"jsonrpc": "2.0",
"id": 2,
"method": "tools/call",
"params": {
"name": "search_web",
"arguments": {
"query": "latest news on MCP protocol"
}
}
}
服务器会返回结果：
{
"jsonrpc": "2.0",
"id": 2,
"result": {
"content": [
{
"type": "text",
"text": "Anthropic released MCP in Nov 2024 as an open standard..."
}
]
}
}
通过实际构建来学习效果最佳，因此接下来的内容将介绍一个被智能体调用的小型MCP服务器。
FastMCP的功能
FastMCP是一个Python框架，可帮助开发者以最少的代码实现MCP服务器。没有它，开发者就必须自行实现JSON-RPC消息循环，对每个请求和响应进行序列化与反序列化，为每个公开函数手动编写JSON架构，并处理诸如initialize、tools/list、tools/call、resources/list和prompts/list之类的协议消息。而使用该框架后，只需一个普通的Python函数加上装饰器即可，其余工作均由框架处理：
from fastmcp import FastMCP
mcp = FastMCP("My Server") # creates the server
@mcp.tool() # registers the function as an MCP tool
def add(a: float, b: float) -> float:
"""Add two numbers.""" # docstring → tool description sent to the LLM
return a + b # type hints → JSON Schema sent to the LLM
mcp.run() # starts the stdio message loop
FastMCP会根据类型注解自行生成JSON模式，而文档字符串则充当模型在判断是否调用某个工具时会读取的通俗描述。调用mcp.run()会启动一个循环，该循环通过stdin和stdout进行通信并使用MCP协议格式进行数据传输，而MultiServerMCPClient正是连接到这个循环上的。
示例服务器中的三种基本操作
项目中的每个服务器都是一个独立的FastMCP应用程序，它们都提供了这三种基本操作类型。
工具是首要考虑的要素：即模型被允许调用的函数。只需用 @mcp.tool() 注解 Python 函数即可，FastMCP 会将该注解及文档字符串转换为 JSON 模式，在 tools/list 请求中发送给客户端。模型通过分析该模式来确定何时使用对应工具以及需要传递哪些参数。
# servers/math_server.py
@mcp.tool()
def divide(a: float, b: float) -> float:
"""Divide a by b. Raises an error if b is zero."""
if b == 0:
raise ValueError("Cannot divide by zero")
return a / b
资源是存储在 URI 中的只读数据。模型不会自动调用它们，因此必须明确请求才能使用。在这个项目中仅用于演示这一概念，而实际应用中可能包括配置文件、数据库快照或参考文档等。
# servers/math_server.py
@mcp.resource("math://constants")
def get_math_constants() -> str:
"""Common mathematical constants."""
return f"π = {math.pi}\n e = {math.e}\n ..."
@mcp.resource("math://formulas/{category}")
def get_formulas(category: str) -> str:
"""Retrieve mathematical formulas by category (geometry | algebra | statistics)."""
catalog = {
"geometry": (
"Geometry Formulas:\n"
" Circle area: A = π × r²\n"
" Circle circumference: C = 2π × r\n"
" Rectangle area: A = length × width\n"
" Triangle area: A = (base × height) / 2\n"
" Sphere volume: V = (4/3) × π × r³\n"
),
"algebra": (
"Algebra Formulas:\n"
" Quadratic formula: x = (−b ± √(b²−4ac)) / 2a\n"
" Difference of squares: a²−b² = (a+b)(a−b)\n"
" Perfect square: (a+b)² = a²+2ab+b²\n"
" Sum of arithmetic seq: S = n(a₁+aₙ)/2\n"
),
"statistics": (
"Statistics Formulas:\n"
" Mean: μ = Σx / n\n"
" Variance: σ² = Σ(x−μ)² / n\n"
" Std Dev: σ = √(Σ(x−μ)² / n)\n"
" Z-score: z = (x−μ) / σ\n"
),
}
return catalog.get(
category,
f"Unknown category '{category}'. Available: geometry, algebra, statistics",
)
URI math://formulas/{category} 是一个模板。其中的 {category} 部分是路径参数，这使得 math://formulas/geometry 和 math://formulas/algebra 成为由同一功能提供的两个独立资源。
提示语是可重复使用的指令模板。提示语是一种参数化的系统消息，客户端可以请求获取；它也是一种将专家角色或任务指令封装在服务器中的方式，从而使所有连接的客户端都能一致地使用这些指令。
# servers/math_server.py
@mcp.prompt()
def math_tutor(difficulty: str = "intermediate") -> str:
return (
f"You are an expert math tutor for {difficulty}-level students. "
"Break every problem into numbered steps..."
)
如下文所示，这些服务器共同提供了19种工具。
客户端的视角
以下是运行带有查询参数的 python main.py 后的具体流程。完整的请求路径如下：
You type a query
│
▼
main.py ← entry point, parses args, kicks off async loop
│
▼
client/agent.py ← spawns 3 MCP servers, builds the agent, invokes it
│ │
│ ▼
│ utils/tracker.py ← fires on every LLM call, tool call, and result
│
▼
LangGraph ReAct loop ← think → call tool → observe → repeat
│
▼
servers/{math,text,data}_server.py ← each runs as an isolated subprocess
第一步：执行起点
main.py是Python首先执行的文件。它会加载环境变量，解析命令行参数，显示欢迎界面，然后将控制权交给异步运行器。
async def _run(queries: list) -> None:
from client.agent import run_query # imported here (late) to keep startup fast
for i, q in enumerate(queries):
await run_query(q)
def main() -> None:
...
asyncio.run(_run(queries))
asyncio.run()会创建一个新的事件循环，并一直阻塞直到_run()执行完毕。演示中使用的查询语句为：
Generate 8 random numbers between 5 and 50 using seed=42,
then calculate their statistics, and tell me if there are any outliers.
通过命令行如下方式运行：
python3 main.py --query "Generate 8 random numbers between 5 and 50 using seed=42, \
then calculate their statistics, and tell me if there are any outliers."
随后通过 _print_header() 打印静态信息面板：
╭────────────────────────────── 🔌 MCP Example ───────────────────────────────╮
│ Multi-Server MCP Demo │
│ │
│ Three FastMCP servers, each exposing tools + resources + prompts: │
│ ● Math Server — add, subtract, multiply, divide, power, sqrt, │
│ percentage │
│ ● Text Server — count_words, word_frequency, reverse, transform, │
│ extract_emails │
│ ● Data Server — generate_numbers, calculate_statistics, find_outliers, │
│ normalise │
│ │
│ Stack : FastMCP · LangChain · LangGraph · OpenAI · Rich │
│ Track : live Rich panels + JSONL log files under logs/ │
╰──────────────────────────────────────────────────────────────────────────────╯
第2步：构建智能体
client/agent.py 中包含用于协调所有操作的异步函数 run_query(query)。该函数首先创建一个跟踪器和一个日志文件。
log_file = str(LOGS_DIR / f"run_{int(time.time())}.jsonl")
tracker = MCPTracker(log_file=log_file)
在连接任何服务器之前，会先创建一个 MCPTracker，并在 logs/ 目录下准备一个带有时间戳的 JSONL 文件。从那时起，所有事件都会实时写入该文件。
接着会打印出用户查询内容：
╭──── 💬 USER QUERY ────╮
│ Generate 8 random... │
╰────────────────────────╯
接着定义需要生成的服务器：
def _server_config() -> Dict[str, Any]:
silent_env = {**os.environ, "FASTMCP_LOG_LEVEL": "ERROR"}
return {
"math_server": {
"command": sys.executable,
"args": ["servers/math_server.py"],
"transport": "stdio",
"env": silent_env,
},
"text_server": { ... },
"data_server": { ... },
}
每条记录描述一个作为子进程运行的MCP服务器。command字段为正在运行的解释器（即sys.executable），args字段指定服务器脚本的名称，transport字段则说明客户端与服务器之间的通信方式，其中stdio表示标准输入输出管道。env字段用于设置子进程的环境，而FASTMCP_LOG_LEVEL=ERROR可隐藏FastMCP的启动提示信息。
MCP中最简单的通信方式是stdio传输方式：客户端将JSON-RPC消息写入服务器的stdin，再从服务器的stdout读取回复，因此无需打开任何网络端口。
之后，客户端会连接到每个服务器并发现其中的工具：
client = MultiServerMCPClient(_server_config())
tools = await client.get_tools()
MultiServerMCPClient 来自 langchain-mcp-adapters，它会同时对每个服务器条目执行以下操作：生成一个子进程，通过标准输入输出发送 MCP initialize 握手信号，发送 tools/list 请求，接收工具架构列表，并将每个架构封装为 LangChain Tool 对象。
当 await client.get_tools() 执行完成时，返回的是包含 19 个 StructuredTool 对象的扁平列表：7 个来自数学服务器，7 个来自文本处理服务器，5 个来自数据处理服务器。这个列表就是模型可使用的全部操作集合。
Connected to 3 servers (math_server, text_server, data_server) with 19 tools: add, subtract, multiply,
divide, power, square_root, calculate_percentage, count_words, word_frequency, reverse_text,
transform_case, find_and_replace, extract_emails, count_vowels_consonants, generate_numbers,
calculate_statistics, find_outliers, sort_values, normalize_values
接着创建语言模型和智能体：
model = ChatOpenAI(model=model_name, temperature=0)
agent = create_react_agent(model, tools)
LangGraph的create_react_agent函数会构建具有如下结构的编译后状态图：
START
│
▼
[call_model] ─── no tool call ──▶ END
│
tool call requested
│
▼
[call_tools]
│
▼
[call_model] (loop again with tool result in context)
这就是ReAct模式，即推理加行动。模型会先对下一步行动进行推理，然后通过调用工具来执行行动，观察结果，如此循环直至获得足够的信息来给出答案。
最后会调用智能体：
config = {"callbacks": [tracker]}
result = await agent.ainvoke({"messages": [("human", query)]}, config=config)
"callbacks": [tracker] 这一配置用于接入 LangChain 的回调系统。每当框架执行重要操作时，比如启动或结束 LLM 调用或工具调用，它就会调用 tracker 中对应的方法。这样，追踪器无需嵌入任何逻辑代码，就能知晓所有发生的事情。
步骤 3：实时观察
MCPTracker 是对 LangChain 的 BaseCallbackHandler 的扩展。其方法永远不会被直接调用，因为 LangChain 会在适当时机自动调用它们。
agent.ainvoke() called
│
├─▶ on_chain_start() "▶ AGENT STARTED" panel
│
├─▶ on_chat_model_start() "🤖 LLM CALL #1" panel (timer starts)
├─▶ on_llm_end() "LLM responded ⏱ 1.23s → will call: add, multiply"
│
├─▶ on_tool_start() "🔧 TOOL CALL #1" panel (timer starts)
├─▶ on_tool_end() "✓ TOOL RESULT ⏱ 0.01s" panel
│
├─▶ on_tool_start() (second tool, if any)
├─▶ on_tool_end()
│
├─▶ on_chat_model_start() "🤖 LLM CALL #2" (LLM synthesises final answer)
├─▶ on_llm_end() no tools this time → loop ends
│
└─▶ agent.ainvoke() returns
每当 on_tool_start 被触发时，追踪器会确定该工具所属的服务器，并为其分配对应的颜色。这样做的唯一目的是提升可读性。
TOOL_SERVER_MAP = {
"add": "math_server",
"count_words": "text_server",
"generate_numbers": "data_server",
...
}
SERVER_COLORS = {
"math_server": "cyan",
"text_server": "green",
"data_server": "yellow",
}
每个事件都会成为 logs/run_<timestamp>.jsonl 文件中的一行 JSON 数据。该文件以追加模式打开并立即写入内容，因此在另一个终端中运行 tail -f logs/run_*.jsonl 可以看到随着代理程序的运行，原始事件内容不断滚动出现。当 agent.ainvoke() 返回后，client/agent.py 会手动调用 print_summary() 函数，该函数会统计整个运行过程中的 LLM 调用次数、工具调用次数以及实际耗时。
首先显示的面板是第 1 次 LLM 调用，其中仅包含一条消息：
── 🤖 LLM CALL #1 model=gpt-5-mini messages=1 ──╮
│ Role Content preview │
│ Human Generate 8 random numbers between 5 and 50… │
╰──────────────────────────────────────────────────────╯
ReAct 图模型刚刚启动。该模型接收由 LangGraph 注入的系统提示，其中包含工具列表及其结构，以及一条人类发送的消息，因此 messages=1。
LLM responded ⏱ 5.35s → will call: generate_numbers
追踪器检查了模型的原始响应。模型并未以文字形式回复，而是以结构化的请求形式要求调用某个工具：
{
"name": "generate_numbers",
"arguments": {"count": 8, "min_val": 5, "max_val": 50, "seed": 42}
}
在 response.generations[0][0].message.tool_calls 中，跟踪器发现了该请求并打印出了其名称。5.35秒仅仅是往返OpenAI API所需的时间。LangGraph读取到这一决策后便进入 call_tools 节点，从而产生第一个工具调用：
╭── 🔧 TOOL CALL #1 ──────────────────╮
│ Tool : generate_numbers │
│ Server: data_server │
│ Args : │
│ {'count': 8, 'min_val': 5, │
│ 'max_val': 50, 'seed': 42} │
╰───────────────────────────────────────╯
从工具名称 generate_numbers 出发，跟踪器查看了 TOOL_SERVER_MAP，该映射对应的值为 data_server，并且将面板颜色设置为黄色，因为 SERVER_COLORS["data_server"] 的值为 "yellow"。就在此时，LangChain 会通过标准输入输出管道向 data_server 子进程发送一条 JSON-RPC tools/call 消息：
{"method": "tools/call", "params": {"name": "generate_numbers",
"arguments": {"count": 8, "min_val": 5, "max_val": 50, "seed": 42}}}
该进程接收输入，运行 Python 函数，然后将结果写回其标准输出。
╭── ✓ TOOL RESULT ⏱ 0.63s ────────────────────────────╮
│ [33.77, 6.13, 17.38, 15.04, 38.14, 35.45, 45.15, 8.91] │
╰──────────────────────────────────────────────────────────╯
它执行了以下精确的代码：
random.seed(42)
return [round(random.uniform(5, 50), 2) for _ in range(8)]
# → [33.77, 6.13, 17.38, 15.04, 38.14, 35.45, 45.15, 8.91]
0.63秒的时间涵盖了从发送请求到通过管道收到响应的整个过程，其中包括Python进程调度的开销。LangGraph会将结果作为ToolMessage添加到智能体的消息列表中，因此该列表现在包含三项内容：人类输入、AI决策（工具调用决策）以及工具输出（结果）。
接着是第2次LLM调用，包含三条消息：
╭── 🤖 LLM CALL #2 model=gpt-5-mini messages=3 ──╮
│ Role Content preview │
│ Human Generate 8 random numbers… │
│ AI (the tool-call decision) │
│ Tool [33.77, 6.13, 17.38, ...] │
╰──────────────────────────────────────────────────────╯
模型会再次使用迄今为止的全部历史数据被调用。它会看到原始问题、自己之前决定调用generate_numbers的记录，以及返回的数字。它的推理过程大致如下：既然已经有了这些数字，而且问题中也要求提供统计信息，那么就应该用这些数字来运行calculate_statistics函数。这样就产生了第二次工具调用及其结果：
Tool : calculate_statistics
Server: data_server
Args : {'numbers': [33.77, 6.13, 17.38, 15.04, 38.14, 35.45, 45.15, 8.91]}
Result: {"count":8, "min":6.13, "max":45.15, "mean":24.9962,
"median":25.575, "std_dev":14.8181, "variance":219.5755,
"range":39.02, "q1":15.04, "q3":38.14}
这里的模型并未记住任何内容。它只是从上下文窗口中先前的ToolMessage中逐字提取数值，然后将它们放入下一个工具的参数中。这就是ReAct循环中链式处理的体现：一个工具的输出会成为后续模型调用可以读取并重复使用的上下文。
在第三次LLM调用中共有五条消息，模型调用了find_outliers函数：
Tool : find_outliers
Server: data_server
Args : {'numbers': [...], 'threshold': 2}
Result: {"outliers": [], "outlier_count": 0,
"total_checked": 8, "mean": 24.9962, "std_dev": 14.8181}
find_outliers函数会计算每个值的z分数（z = (x − mean) / std_dev），并检查是否有值超过2的阈值。由于没有值超出该阈值，因此outliers是一个空列表。
LLM调用编号4共收到了七条消息：
╭── 🤖 LLM CALL #4 messages=7 ──╮
│ Human / AI / Tool │
│ AI / Tool │
│ AI / Tool │
╰───────────────────────────────────╯
LLM responded ⏱ 30.25s
缺失的“will call”后缀表明模型生成的是普通文本而非工具请求，从而结束了ReAct循环。由于上下文量较大（包含七条消息及原始JSON结果），且模型需要手动为每个数值计算z分数，因此整个调用过程耗时30秒。LangGraph在响应中未发现任何工具调用，于是将流程转向END，进而输出最终答案：
╭── ✅ FINAL ANSWER ───────────────────────────────╮
│ Random numbers: [33.77, 6.13, 17.38, ...] │
│ Mean: 24.9962 / Median: 25.575 / Std Dev: 14.8181│
│ Outliers: none (all |z| < 2) │
╰───────────────────────────────────────────────────╯
随后会显示运行摘要：
╭─────────────────────────────────────────────────────╮
│ LLM Calls 4 │
│ Tool Calls 3 │
│ Total Time 42.25s │
│ Log File logs/run_1776864980.jsonl │
╰─────────────────────────────────────────────────────╯
完整的消息序列
这是当 agent.ainvoke() 返回时，代理消息列表中的所有条目：
1 HumanMessage "Generate 8 random numbers…"
2 AIMessage [tool_call: generate_numbers({count:8, min:5, max:50, seed:42})]
3 ToolMessage [33.77, 6.13, 17.38, 15.04, 38.14, 35.45, 45.15, 8.91]
4 AIMessage [tool_call: calculate_statistics({numbers:[...]})]
5 ToolMessage {count:8, mean:24.9962, std_dev:14.8181, ...}
6 AIMessage [tool_call: find_outliers({numbers:[...], threshold:2})]
7 ToolMessage {outliers:[], outlier_count:0, ...}
8 AIMessage "Here are the results. Random numbers: …" ← final answer
每次调用大语言模型时都会使用此前积累的所有数据，因此跟踪面板中的messages数值会以1、3、5、7的步长递增。
JSONL日志
上述所有事件在发生时都会被记录到logs/run_x.jsonl中，每行对应一个JSON对象：
{"timestamp": "...", "event": "agent_start", "data": {"runnable": "LangGraph", "run_id": "..."}}
{"timestamp": "...", "event": "llm_call", "data": {"call_number": 1, "model": "gpt-5-mini", "messages": 1}}
{"timestamp": "...", "event": "llm_end", "data": {"elapsed": "5.35s", "tool_calls_requested": ["generate_numbers"]}}
{"timestamp": "...", "event": "tool_start", "data": {"call_number": 1, "tool": "generate_numbers", "server": "data_server", "args": "..."}}
{"timestamp": "...", "event": "tool_end", "data": {"elapsed": "0.63s", "output_preview": "[33.77, 6.13, ...]"}}
{"timestamp": "...", "event": "llm_call", "data": {"call_number": 2, "model": "gpt-5-mini", "messages": 3}}
...
{"timestamp": "...", "event": "run_summary", "data": {"llm_calls": 4, "tool_calls": 3, "total_elapsed": "42.25s"}}
为何这很重要
MCP的真正价值在于其分布式特性。在标准出现之前，任何希望让大语言模型具备超出其固有能力的功能的团队都不得不自行构建集成层，而这些工作成果无法应用到其他模型或框架上。如今，只需编写一次服务器代码，即可为所有符合标准的客户端提供服务。数据库团队可以为其数据结构发布只读服务器，DevOps团队可以提供部署工具，数据平台则可以提供查询与分析功能。任何智能体都能连接到这些服务，各团队也无需就实现细节达成一致。
相关阅读
- 使用 LangChain Deep Agents 重建 Claude Code 风格的智能体 — 从核心循环到工具、规划、内存、子智能体以及沙箱式审批，详细讲解如何通过 DeepAgents 逐步重建一个编程智能体。
- 开发者优先安装的 10 个 MCP 服务器 — 为 AI 编程助手整理的按优先级排序的 MCP 服务器列表，涵盖 Context7、GitHub、Postgres 等工具，同时提供安装配置及权限设置注意事项。