实用笔记：MCP的工作原理——结合代码的深入解析
《实用笔记：MCP的工作原理——带代码的深入解析》操作指南：为采用该模式的团队提供的契约、校验机制以及可直接插入的代码模块。
本指南将逐步构建从原材料到可运行系统的完整流程，内容为《MCP的工作原理：基于代码的深入解析》。重点在于可操作的步骤、明确的检查点，以及可直接放入代码库而无需猜测其用途的代码。 在概览阶段，应在修改代码之前明确输入参数、各步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新运行相应步骤，而无需猜测隐藏的状态。 除了功能结果外，还需记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在流程从演示环境过渡到共享环境时出现意外费用。
{
"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..."
}
]
}
}
什么是FastMCP？
在学习“什么是FastMCP”这一阶段时，首先需写下相关契约：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 应将配置信息与应用程序代码分开存放。环境文件、密钥存储以及功能开关应集中于一个位置，这样操作人员无需查看整个系统结构即可进行审计。 对于每次调用，都要记录工具名称、参数哈希值、延迟时间以及执行结果。如果没有这些记录，调试代理循环问题将会耗费大量时间。
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
三种MCP基本操作
在处理三个MCP原语阶段时，首先需写明契约内容：所需输入、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 同时记录正常流程和恢复流程。重试机制、人工审核环节以及死信处理都是产品功能的一部分，而非后续的优化工作。 需为每次调用记录工具名称、参数哈希值、延迟时间以及最终结果。没有这些记录，调试代理将陷入无休止的循环，耗费大量时间。
# 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
# 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",
)
# 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..."
)
服务器概要
在处理服务器总结阶段时，首先记下相关约定：所需输入、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 优先选择小型、可测试的单元，而非冗长的脚本。当某个步骤失败时，故障应能指向单一责任模块，而非复杂的流程链。 需为每次调用记录工具名称、参数哈希值、延迟时间以及最终结果。没有这些记录，调试代理将陷入无休止的循环，耗费大量时间。
从客户端的角度看
在处理“来自客户端”阶段时，首先写下合同条款：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 将此阶段视为输入与验证后输出之间的契约。为相关成果命名，明确成功判定标准，杜绝无声的半完成状态。 需记录每次调用的工具名称、参数哈希值、延迟时间以及最终结果。没有这些记录，调试代理将陷入无休止的循环，浪费大量时间。
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
第一步——入口点
在完成第一步的入门阶段时，首先写下合同规范：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。 为每次调用记录工具名称、参数哈希值、延迟时间以及最终结果。没有这些记录，调试代理循环将会浪费大量时间。
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))
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."
╭────────────────────────────── 🔌 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/ │
╰──────────────────────────────────────────────────────────────────────────────╯
第二步 — 构建代理
在执行“构建阶段”的第二步时，首先写下相关契约：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 将配置信息置于应用程序代码之外。环境文件、密钥存储以及功能开关应集中存放于一个位置，这样操作人员无需查看整个系统结构即可进行审计。 为每次调用记录工具名称、参数哈希值、延迟时间以及执行结果。如果没有这些记录，调试代理将陷入无止境的循环，耗费大量时间。
log_file = str(LOGS_DIR / f"run_{int(time.time())}.jsonl")
tracker = MCPTracker(log_file=log_file)
╭──── 💬 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": { ... },
}
client = MultiServerMCPClient(_server_config())
tools = await client.get_tools()
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)
START
│
▼
[call_model] ─── no tool call ──▶ END
│
tool call requested
│
▼
[call_tools]
│
▼
[call_model] (loop again with tool result in context)
config = {"callbacks": [tracker]}
result = await agent.ainvoke({"messages": [("human", query)]}, config=config)
第三步 — 实时监控一切
在完成第三步“观察所有情况”阶段时，首先写下相关契约：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及死信处理都是产品功能的一部分，而非后续需要补充的内容。 需为每次调用记录工具名称、参数哈希值、延迟时间以及最终结果。没有这些记录，调试代理将陷入无止境的循环，耗费大量时间。
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
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",
}
── 🤖 LLM CALL #1 model=gpt-5-mini messages=1 ──╮
│ Role Content preview │
│ Human Generate 8 random numbers between 5 and 50… │
╰──────────────────────────────────────────────────────╯
LLM responded ⏱ 5.35s → will call: generate_numbers
{
"name": "generate_numbers",
"arguments": {"count": 8, "min_val": 5, "max_val": 50, "seed": 42}
}
╭── 🔧 TOOL CALL #1 ──────────────────╮
│ Tool : generate_numbers │
│ Server: data_server │
│ Args : │
│ {'count': 8, 'min_val': 5, │
│ 'max_val': 50, 'seed': 42} │
╰───────────────────────────────────────╯
{"method": "tools/call", "params": {"name": "generate_numbers",
"arguments": {"count": 8, "min_val": 5, "max_val": 50, "seed": 42}}}
╭── ✓ 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]
╭── 🤖 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, ...] │
╰──────────────────────────────────────────────────────╯
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}
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}
╭── 🤖 LLM CALL #4 messages=7 ──╮
│ Human / AI / Tool │
│ AI / Tool │
│ AI / Tool │
╰───────────────────────────────────╯
LLM responded ⏱ 30.25s
╭── ✅ 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 │
╰─────────────────────────────────────────────────────╯
完整的消息序列
在处理“完整消息序列”阶段时，首先需明确契约内容：所需输入、成功信号以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 优先选择小型、可测试的单元，而非冗长的脚本。当某一步骤失败时，故障应指向单一责任点，而非复杂的流程链。 需为每次调用记录工具名称、参数哈希值、延迟时间以及最终结果。没有这些记录，调试过程将会浪费大量时间。 在处理“完整消息序列”阶段时，首先需明确契约内容：所需输入、成功信号以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 在功能结果旁还需记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。
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
JSONL日志
将JSONL日志阶段视为可度量的对象来处理时，其效果最佳。在扩大范围之前，先记录一个成功的示例、一个失败案例以及回滚说明。 应将配置与应用程序代码分开。环境文件、密钥存储和功能标志应集中存放于一个位置，以便操作人员无需查看整个系统结构即可进行审计。 提供具有明确架构和清晰副作用标签的工具。主机需要在自动批准之前知道哪些调用会修改状态。
{"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"}}
来源
将“来源”阶段视为可度量的对象来处理，效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。同时记录正常流程和恢复流程。重试机制、人工审核环节以及死信处理都是产品本身的一部分，而非后续需要补充的内容。应提供具有明确结构规范和清晰副作用标识的工具，这样主机在自动批准之前就能知道哪些调用会改变状态。
来自创始人的讯息
将我们测试阶段的“A消息”视为可度量的对象来处理效果最佳。在扩大范围之前，先记录一份理想的执行日志、一个故障案例以及回滚说明。 相较于庞大的脚本，应优先选择小型且可测试的单元。当某个步骤出现故障时，故障点应指向单一责任模块，而非复杂的流程链。 使用结构明确的工具，并为各种操作添加清晰的副作用标签。主机需要在自动批准之前知道哪些调用会修改状态。 将我们测试阶段的“A消息”视为可度量的对象来处理效果最佳。在扩大范围之前，先记录一份理想的执行日志、一个故障案例以及回滚说明。 除了功能结果外，还需记录执行时间以及令牌或查询的成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。
操作检查清单
将操作检查清单阶段视为可度量的工作面，效果最佳。在扩大范围之前，先记录一份完美的测试用例、一个故障案例以及回滚说明。
把这一阶段视为输入与已验证输出之间的契约。为相关成果命名，明确成功标准，绝不允许默许不完整的处理结果。
使用具有严格结构定义和明确副作用标注的工具。主机需要在自动批准之前知道哪些调用会改变系统状态。
在预算允许的情况下，利用测试环境而非真实的付费 API，在持续集成过程中添加用于检测关键路径的冒烟测试。
在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免从演示环境过渡到共享环境时出现意外账单。
应公开那些具有有限架构且带有明确副作用标签的工具。主机在自动批准之前需要知道哪些调用会修改状态。
在升级技术栈之前，需冻结版本、为关键路径记录完整日志，并确认回滚步骤。共享环境需要设置速率限制、进行租户验证，同时明确密钥轮换的负责人。与其追求花哨的一次性演示，不如注重扎实的可靠性。
c7efc4f69698的批量说明：请将提供商密钥存放在仓库之外，设定单会话令牌上限，并将日志与评估用文件一起存储，以便后续模型更换时仍能保持对比性。
在进入强化措施的第0阶段时，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新执行该步骤，而无需猜测隐藏状态。需同时记录正常流程与异常恢复流程。重试机制、人工审核环节以及错误处理方式都是产品本身的组成部分，而非后续需要补充的内容。
强化措施细节0/888：需统计该措施的耗时、错误类型以及令牌使用情况，然后依据固定的评估标准而非个人经验来决定是否保留该变更。
在处理强化措施的第一阶段时，首先写下相关契约：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改始终符合约定。 将这一阶段视为输入与验证后输出之间的契约。为相关产物命名，明确成功判定标准，杜绝默许的部分完成情况。
强化措施细节 1/888：针对该事项记录执行耗时、错误类型以及令牌消耗情况，然后依据固定的评估标准而非个人经验来决定是否保留该变更。
将强化措施的第二阶段视为可度量的对象来处理效果最佳。在扩大范围之前，先记录一份标准操作示例、一个失败案例以及回滚说明。 应将配置信息与应用程序代码分开。环境文件、密钥存储和功能开关应集中存放于一处，以便操作人员无需查看全部代码结构即可进行审计。
强化措施细节 2/888：记录该任务的执行时间、错误类型以及代币消耗情况，然后依据固定的评估标准而非个人经验来决定是否保留该变更。
在强化措施的第3阶段，应在修改代码之前明确输入参数、该步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏的状态。相比复杂的脚本，更应采用小型且可测试的单元。当某一步骤失败时，故障原因应能明确指向某个具体的责任方，而非整个混乱的流程。
强化措施细节 3/888：记录该任务的执行时间、错误类型以及代币消耗情况，然后依据固定的评估标准而非个人经验来决定是否保留该变更。