作 者:吴佳浩(Alben)
公众号:全栈架构师笔记
系列专栏:《MCP 与 Agent Tools 工程化落地实战》· 第 01 篇
导读
Function Calling 只是单次 HTTP 请求的语法糖,MCP(Model Context Protocol)才是 AI 时代的标准化 USB 总线。
以前每个 Agent 框架都在重复造工具适配器的轮子,写一套代码只能给一个框架用;MCP 第一次把 Tools、Resources 和 Prompts 变成了跨模型、跨框架、跨语言的通用基础设施。
函数调用解决了“模型怎么输出参数”,MCP 解决了“智能体如何与世界解耦交互”。
在过去两年中,几乎所有做 Agent 的团队都经历过这样一段痛苦的技术演进: 早期 OpenAI 推出了 Function Calling,大家兴奋地在业务代码里手写一个个 JSON Schema。但随着接入的外部系统越来越多——内部数据库、GitLab、Jira、Kubernetes、飞书文档、本地文件系统——团队会立刻撞上一面坚硬的工程之墙:
| 困境现象 | 具体表现 | 架构根因 |
|---|---|---|
| 1. 碎片化与厂商锁定 | 为 LangChain 写的工具,在 | 缺乏统一的协议标准,每个框架 |
| (Vendor Lock-in) | AutoGen 或 Claude Code 里跑不通 | 都有自己私有的 Tool 抽象基类 |
| 2. 状态与连接管理缺失 | 每次函数调用都是无状态短连接, | 协议层缺乏长会话生命周期维护与 |
| (Stateless Churn) | 无法支持流式推送、事件订阅与鉴权 | 双向通信通道 |
| 3. 资源与上下文混淆 | 文档、日志、表格等静态资源,全被 | 缺乏资源与操作的语义正交解耦, |
| (Semantic Pollution) | 强行包成函数,导致模型意图混乱 | 导致 Prompt 空间极度低效 |
为了解决这种混乱的“工具孤岛”局面,Anthropic 开源了 MCP(Model Context Protocol),并在短短几个月内迅速成为事实上的行业标准。
为什么行业在有了 Function Calling 之后,依然迫切需要 MCP?MCP 底层究竟设计了哪些精妙的机制?
一、从 Function Calling 到 MCP:协议化演进的必然性
要理解 MCP 的价值,我们必须看清工具调用在架构上的四代演进:
- 🔸 第一代(Prompt 裸搓):解析经常失败,极不稳定;
- 🔸 第二代(Function Calling):模型保证了 JSON 结构的稳定性,但它只是一套 API 序列化协议,完全不涉及工具在哪里运行、如何鉴权、如何跨网络发现;
- 🔸 第三代(框架私有 SDK):形成了严重的框架壁垒,团队在工具维护上浪费了大量无意义的胶水代码;
- 🔸 第四代(MCP 标准协议):彻底将 Client(宿主应用/Agent) 与 Server(工具与数据提供方) 解耦。无论底层是大模型 A 还是大模型 B,只要支持 MCP,就能即插即用接入全球所有的 MCP Servers。
一句话总结这一章的核心观点:
Function Calling 只是模型接口层面的特性,MCP 则是整个分布式智能体生态的通信协议。
二、MCP 核心三要素:Tools、Resources 与 Prompts 的正交设计
很多初学者把 MCP 简单理解为“远程函数调用(RPC)”。这是极其片面的。MCP 协议的核心精髓,在于它将上下文交互正交拆解为三大支柱:
| 核心要素 | 抽象定位 | 交互模式 | 典型应用场景 |
|---|---|---|---|
| 🛠️ Tools (工具) | 模型可调用的动作 | 动态调用 (Model-Pull) | 执行 SQL、部署容器、 |
| (具备副作用) | 需大模型主动下发参数 | 修改本地文件、调 API | |
| 📄 Resources (资源) | 模型可读取的数据 | 被动装载 (App/User) | 读取日志、查看表结构、 |
| (只读无副作用) | 类似文件或 URI 数据源 | 获取 Git Diff、系统指标 | |
| 💬 Prompts (提示词) | 预定义的交互模板 | 显式触发 (User-Push) | 单元测试生成模板、 |
| (工程化工作流) | 固化的专家级提问范式 | 代码 Review 标准规程 |
- 🔸 Tools(动作):赋予 Agent 改变世界的能力(有副作用,必须受权限与审批管控);
- 🔸 Resources(数据):赋予 Agent 观察世界的能力(标准化 URI 寻址,如
git://repo/diff或db://schema/users,只读且幂等); - 🔸 Prompts(模版):固化了人类专家的交互最佳实践,让用户一键激活复杂的多步指令。
一句话总结这一章的核心观点:
Tools 是手,Resources 是眼,Prompts 是任务书。三者解耦,才构成了完整的上下文交互协议。
三、MCP 通信管道:Stdio 与 Streamable HTTP/SSE 的选型权衡
在工程实现上,MCP 支持两种底层传输通道(Transports),它们适用于完全不同的物理场景:
| 对比维度 | Stdio Transport (标准输入输出) | Streamable HTTP / SSE |
|---|---|---|
| 通信机制 | 进程间管道 (stdin / stdout) | 长连接 HTTP + Server-Sent-Evt |
| 部署拓扑 | 本地子进程 (Subprocess) | 跨网络分布式服务 (Microservice |
| 鉴权与隔离 | 依赖操作系统进程级权限 | OAuth2、JWT、mTLS 标准网关 |
| 典型场景 | 桌面客户端、CLI 工具、本地排错 | 企业中台、多 Agent 共享微服务 |
| 优势与代价 | 零网络开销、极速冷启动; | 跨机器共享、弹性伸缩; |
| 无法跨机器共享、调试困难 | 需维护长连接与分布式网关 |
一句话总结这一章的核心观点:
本地极客工具选 Stdio 极速启动,企业级中台必须上 Streamable HTTP/SSE 实现多租户鉴权与共享。
四、生产级代码实战:手写一个标准 MCP Client
以下为基于 Python 3.11+ 与标准库构建的极简、无第三方重型依赖的 MCP Stdio Client 核心实现:
<span>"""
mcp_client_runtime.py
生产级轻量 MCP Client 实现
支持:
- Stdio 子进程生命周期管理
- MCP Initialize 能力协商
- Tool Schema 获取
- Tool 调用
"""</span>
<span>import</span> json
<span>import</span> subprocess
<span>import</span> threading
<span>from</span> typing <span>import</span> <span>Any</span>, <span>Dict</span>, <span>List</span>, <span>Optional</span>
<span>class</span> <span>MCPStdioClient</span>:
<span>"""基于 Stdio 的 MCP Client"""</span>
<span>def</span> <span>__init__</span>(<span>self, command: <span>str</span>, args: <span>List</span>[<span>str</span>]</span>):
self.command = command
self.args = args
self.process: <span>Optional</span>[subprocess.Popen] = <span>None</span>
self._request_id = <span>0</span>
self._lock = threading.Lock()
<span>def</span> <span>start</span>(<span>self</span>) -> <span>None</span>:
<span>"""启动 MCP Server 并完成 Initialize 握手"""</span>
self.process = subprocess.Popen(
[self.command] + self.args,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=<span>True</span>,
bufsize=<span>0</span>,
)
<span># Initialize</span>
self._send_raw(
{
<span>"jsonrpc"</span>: <span>"2.0"</span>,
<span>"id"</span>: self._next_id(),
<span>"method"</span>: <span>"initialize"</span>,
<span>"params"</span>: {
<span>"protocolVersion"</span>: <span>"2024-11-05"</span>,
<span>"capabilities"</span>: {},
<span>"clientInfo"</span>: {
<span>"name"</span>: <span>"FullStackArchitectClient"</span>,
<span>"version"</span>: <span>"1.0.0"</span>,
},
},
}
)
self._read_response()
<span># Initialized Notification</span>
self._send_raw(
{
<span>"jsonrpc"</span>: <span>"2.0"</span>,
<span>"method"</span>: <span>"notifications/initialized"</span>,
}
)
<span>def</span> <span>list_tools</span>(<span>self</span>) -> <span>List</span>[<span>Dict</span>[<span>str</span>, <span>Any</span>]]:
<span>"""获取 MCP Server 暴露的 Tool Schema"""</span>
req_id = self._next_id()
self._send_raw(
{
<span>"jsonrpc"</span>: <span>"2.0"</span>,
<span>"id"</span>: req_id,
<span>"method"</span>: <span>"tools/list"</span>,
<span>"params"</span>: {},
}
)
response = self._read_response()
<span>return</span> response.get(<span>"result"</span>, {}).get(<span>"tools"</span>, [])
<span>def</span> <span>call_tool</span>(<span>
self,
tool_name: <span>str</span>,
arguments: <span>Dict</span>[<span>str</span>, <span>Any</span>],
</span>) -> <span>str</span>:
<span>"""调用远程 MCP Tool"""</span>
req_id = self._next_id()
self._send_raw(
{
<span>"jsonrpc"</span>: <span>"2.0"</span>,
<span>"id"</span>: req_id,
<span>"method"</span>: <span>"tools/call"</span>,
<span>"params"</span>: {
<span>"name"</span>: tool_name,
<span>"arguments"</span>: arguments,
},
}
)
response = self._read_response()
content = response.get(<span>"result"</span>, {}).get(<span>"content"</span>, [])
<span>return</span> <span>"\n"</span>.join(
item.get(<span>"text"</span>, <span>""</span>)
<span>for</span> item <span>in</span> content
<span>if</span> item.get(<span>"type"</span>) == <span>"text"</span>
)
<span>def</span> <span>_next_id</span>(<span>self</span>) -> <span>int</span>:
<span>"""生成递增 Request ID"""</span>
<span>with</span> self._lock:
self._request_id += <span>1</span>
<span>return</span> self._request_id
<span>def</span> <span>_send_raw</span>(<span>self, payload: <span>Dict</span>[<span>str</span>, <span>Any</span>]</span>) -> <span>None</span>:
<span>"""发送 JSON-RPC 请求"""</span>
<span>if</span> <span>not</span> self.process <span>or</span> <span>not</span> self.process.stdin:
<span>raise</span> RuntimeError(<span>"MCP process has not been started."</span>)
raw = json.dumps(payload) + <span>"\n"</span>
self.process.stdin.write(raw)
self.process.stdin.flush()
<span>def</span> <span>_read_response</span>(<span>self</span>) -> <span>Dict</span>[<span>str</span>, <span>Any</span>]:
<span>"""读取 JSON-RPC 响应"""</span>
<span>if</span> <span>not</span> self.process <span>or</span> <span>not</span> self.process.stdout:
<span>raise</span> RuntimeError(<span>"MCP process has not been started."</span>)
line = self.process.stdout.readline()
<span>if</span> <span>not</span> line:
stderr = <span>""</span>
<span>if</span> self.process.stderr:
stderr = self.process.stderr.read()
<span>raise</span> RuntimeError(
<span>f"MCP Server closed unexpectedly: <span>{stderr}</span>"</span>
)
<span>return</span> json.loads(line.strip())
<span>def</span> <span>close</span>(<span>self</span>) -> <span>None</span>:
<span>"""关闭 MCP Server"""</span>
<span>if</span> self.process:
self.process.terminate()
self.process.wait()
self.process = <span>None</span>
本篇总结
- 🔸 Function Calling 只是语法糖,MCP 才是 AI 外设生态的标准化 USB 总线;
- 🔸 Tools、Resources、Prompts 三要素正交解耦,构成了完整的上下文交互规范;
- 🔸 Stdio 适合本地,Streamable HTTP/SSE 适合企业级中台;
- 🔸 掌握协议层,才能真正构建可插拔、可演进的工业级 Agent。
在下一篇中,我们将深入实战:《从零手写一个生产级 MCP Server:鉴权、流式传输与状态管理》,带你从零构建一个高可用的企业级数据连接器!
筒子们本篇为《企业级 Agent 实战指南》· 第二章的第 01 篇,后续续会更新完整的agent的开发的全部过程,如果你对Agent开发感兴趣不妨关注一下本合集。
适合正在选型 Agent 工具链的团队阅读,能快速建立协议层认知,避免重复造适配器轮子,为后续工程化落地打好基础。