从源码角度深度剖析MCP协议：AI工具调用标准化革命
·
从源码角度深度剖析MCP协议：AI工具调用的「USB-C」时刻来了
作者说： 2025年，Anthropic扔出了MCP（Model Context Protocol）协议，宣称要让AI工具调用像USB-C一样即插即用。这不是营销话术——从源码拆解来看，MCP确实解决了一个真实痛点。本文从协议设计、源码实现、实战踩坑三个维度，把MCP说透。
目录
- 1. 背景：为什么AI工具调用这么痛苦
- 2. MCP协议架构解析
- 3. 源码级协议实现拆解
- 4. 实战：5分钟跑通MCP Server
- 5. 生产级实战：构建RAG检索MCP Server
- 6. 当前局限性与未来方向
- 7. 总结与参考资料
1. 背景：为什么AI工具调用这么痛苦
1.1 工具调用范式的演进
在MCP出现之前，AI Agent调用外部工具经历了三个阶段：
第一阶段：Function Calling（2023）
├── 每个模型有自己的function calling格式
├── OpenAI用function call，Claude用tool use
└── 问题：格式不通用，换模型要重写
第二阶段：Tool Use协议（2023-2024）
├── 各平台逐步标准化
├── JSON Schema定义工具签名
└── 问题：每个工具集都要独立适配，没有统一传输层
第三阶段：MCP协议（2025）
├── 统一的工具发现 + 调用 + 传输协议
├── 服务端/客户端分离架构
└── 核心解决：一次实现，到处调用
1.2 MCP解决的核心问题
MCP解决的不是「如何让AI调用工具」，而是**「如何让工具以一种标准方式被任何AI调用」**。
没有MCP时：
AI模型A → [专有适配器] → 工具X
AI模型B → [专有适配器] → 工具X ← 重写！
AI模型C → [专有适配器] → 工具X ← 又要重写！
有MCP后：
工具X → [MCP Server] → MCP Client（通用）→ AI模型A/B/C
MCP的本质：把「工具适配」变成一次性的基础设施工作。
2. MCP协议架构解析
2.1 协议分层
MCP采用三层协议架构：
┌─────────────────────────────────────┐
│ Application Layer（应用层） │ ← AI模型理解的部分
│ prompts / resources / tools │
├─────────────────────────────────────┤
│ Protocol Layer（协议层） │ ← JSON-RPC 2.0消息格式
│ JSON-RPC 2.0 + MCP Message Types │
├─────────────────────────────────────┤
│ Transport Layer（传输层） │ ← 支持stdio / HTTP(SSE)
│ stdio / HTTP + SSE │
└─────────────────────────────────────┘
2.2 核心消息类型
MCP定义了三类核心能力，通过JSON-RPC 2.0协议传输：
// 1. 工具调用请求（Client → Server）
{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": "filesystem_read_file",
"arguments": {
"path": "/data/config.json"
}
}
}
// 2. 工具列表响应（Server → Client）
{
"jsonrpc": "2.0",
"id": 2,
"result": {
"tools": [
{
"name": "filesystem_read_file",
"description": "读取文件内容",
"inputSchema": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "文件路径"
}
},
"required": ["path"]
}
}
]
}
}
// 3. 资源访问请求（Client → Server）
{
"jsonrpc": "2.0",
"id": 3,
"method": "resources/list",
"params": {}
}
2.3 传输层：stdio vs HTTP
MCP支持两种传输方式，适用于不同场景：
| 传输方式 | 适用场景 | 优点 | 缺点 |
|---|---|---|---|
| stdio | 本地进程通信（Claude Desktop等） | 简单、安全、易调试 | 不适合远程 |
| HTTP+SSE | 远程服务 | 支持分布式部署 | 需要处理连接管理 |
# stdio模式：子进程启动
# MCP Server作为子进程，通过stdin/stdout通信
# HTTP+SSE模式：远程服务
# MCP Server作为HTTP服务，Client通过HTTP请求通信
3. 源码级协议实现拆解
3.1 MCP SDK核心类图
以官方Python SDK（mcp-sdk）为例，核心类的设计：
# mcp/server/models.py（简化版核心模型）
class Tool:
"""工具定义"""
def __init__(
self,
name: str, # 工具唯一标识
description: str, # AI模型用于理解工具用途
inputSchema: dict, # JSON Schema格式的参数定义
):
self.name = name
self.description = description
self.inputSchema = inputSchema
class Resource:
"""资源定义 - 供AI读取的上下文数据"""
def __init__(
self,
uri: str, # 资源唯一标识（格式：scheme://path）
name: str,
description: str = None,
mimeType: str = "text/plain",
):
self.uri = uri
self.name = name
self.description = description
self.mimeType = mimeType
class Prompt:
"""提示模板 - AI可调用的提示工程组件"""
def __init__(
self,
name: str,
description: str,
arguments: list[PromptArgument] = None, # 可选参数
):
self.name = name
self.description = description
self.arguments = arguments or []
3.2 Server生命周期源码解析
# mcp/server/server.py（核心逻辑简化）
class MCPServer:
"""MCP Server主类"""
def __init__(self, name: str):
self.name = name
self._tools: dict[str, Tool] = {}
self._resources: dict[str, Resource] = {}
self._prompts: dict[str, Prompt] = {}
self._request_handlers: dict[str, Callable] = {}
# 注册内置的协议方法处理器
self._register_protocol_handlers()
def _register_protocol_handlers(self):
"""MCP协议规定必须支持的6个基础方法"""
self._request_handlers["initialize"] = self._handle_initialize
self._request_handlers["tools/list"] = self._handle_tools_list
self._request_handlers["tools/call"] = self._handle_tools_call
self._request_handlers["resources/list"] = self._handle_resources_list
self._request_handlers["resources/read"] = self._handle_resources_read
self._request_handlers["prompts/list"] = self._handle_prompts_list
async def _handle_initialize(self, params: dict) -> dict:
"""
握手初始化 - MCP协议的第一步
Client发送AI模型的能力，Server回复自己支持的能力
"""
client_info = params.get("clientInfo", {})
protocol_version = params.get("protocolVersion", "2024-11-05")
return {
"protocolVersion": protocol_version,
"serverInfo": {
"name": self.name,
"version": "1.0.0"
},
"capabilities": {
"tools": {"listChanged": True}, # 声明支持工具变更通知
"resources": {"subscribe": True, "listChanged": True},
"prompts": {"listChanged": True}
}
}
async def _handle_tools_list(self, params: dict) -> dict:
"""返回所有可用工具"""
return {
"tools": [
{
"name": tool.name,
"description": tool.description,
"inputSchema": tool.inputSchema
}
for tool in self._tools.values()
]
}
async def _handle_tools_call(self, params: dict) -> dict:
"""
工具调用 - 核心方法
AI模型通过这个方法真正执行工具逻辑
"""
tool_name = params["name"]
arguments = params.get("arguments", {})
if tool_name not in self._tools:
raise McpError(f"Unknown tool: {tool_name}")
tool = self._tools[tool_name]
# 参数验证（基于JSON Schema）
self._validate_arguments(tool.inputSchema, arguments)
# 调用工具处理函数
handler = self._tool_handlers[tool_name]
result = await handler(**arguments)
return {
"content": [
{
"type": "text",
"text": json.dumps(result, ensure_ascii=False)
}
],
"isError": False
}
3.3 JSON Schema验证逻辑
工具调用的参数验证是MCP的核心安全机制：
# mcp/server/validation.py（参数验证核心逻辑）
def validate_input_schema(schema: dict, arguments: dict) -> list[str]:
"""
验证工具调用参数是否符合inputSchema定义
返回错误信息列表，空列表表示验证通过
"""
errors = []
# 检查必填参数
required = schema.get("required", [])
for field in required:
if field not in arguments:
errors.append(f"Missing required field: {field}")
# 检查每个提供的参数
properties = schema.get("properties", {})
for key, value in arguments.items():
if key not in properties:
errors.append(f"Unknown field: {key}")
continue
prop_schema = properties[key]
type_expected = prop_schema.get("type")
# 类型检查
if type_expected and not _check_type(value, type_expected):
errors.append(
f"Field '{key}' expected type '{type_expected}', "
f"got '{type(value).__name__}'"
)
# 枚举值检查
if "enum" in prop_schema and value not in prop_schema["enum"]:
errors.append(
f"Field '{key}' must be one of {prop_schema['enum']}, "
f"got '{value}'"
)
return errors
4. 实战：5分钟跑通MCP Server
4.1 环境准备
# 安装MCP Python SDK
pip install mcp
# 验证安装
python -c "import mcp; print(mcp.__version__)"
4.2 最简MCP Server：5步创建
完整代码（server.py）：
#!/usr/bin/env python3
"""
MCP Server 极简示例：文件搜索工具
功能：AI可以通过此Server搜索本地文件内容
"""
import mcp.server.stdio
import mcp.types as types
from pathlib import Path
import json
import asyncio
class FileSearchServer:
"""文件搜索MCP Server"""
def __init__(self):
self.search_history = [] # 记录搜索历史
# ========== 工具定义 ==========
def get_tools(self) -> list[types.Tool]:
"""定义Server提供的工具"""
return [
types.Tool(
name="file_search",
description=(
"在指定目录中搜索包含关键词的文件。"
"支持TXT、MD、JSON、PY文件类型。"
"返回匹配文件的路径和行号。"
),
inputSchema={
"type": "object",
"properties": {
"directory": {
"type": "string",
"description": "要搜索的目录路径"
},
"keyword": {
"type": "string",
"description": "搜索关键词（支持正则）"
},
"file_type": {
"type": "string",
"description": "文件类型过滤，如 'py', 'md'",
"enum": ["py", "md", "txt", "json", "all"],
"default": "all"
},
"max_results": {
"type": "integer",
"description": "最大返回结果数",
"default": 20,
"minimum": 1,
"maximum": 100
}
},
"required": ["directory", "keyword"]
}
),
types.Tool(
name="file_read",
description="读取文件内容并返回指定行范围",
inputSchema={
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "文件完整路径"
},
"start_line": {
"type": "integer",
"description": "起始行号（1-based）",
"default": 1
},
"end_line": {
"type": "integer",
"description": "结束行号（含）",
"default": 100
}
},
"required": ["path"]
}
)
]
# ========== 工具处理逻辑 ==========
async def handle_tool_call(
self,
name: str,
arguments: dict
) -> list[types.ContentBlock]:
"""工具调用入口"""
if name == "file_search":
return await self._handle_file_search(**arguments)
elif name == "file_read":
return await self._handle_file_read(**arguments)
else:
raise ValueError(f"Unknown tool: {name}")
async def _handle_file_search(
self,
directory: str,
keyword: str,
file_type: str = "all",
max_results: int = 20
) -> list[types.ContentBlock]:
"""执行文件搜索"""
import re
dir_path = Path(directory)
if not dir_path.exists():
return [types.TextContent(
text=json.dumps({"error": f"目录不存在: {directory}"}, ensure_ascii=False)
)]
# 确定搜索的文件类型
extensions = {
"py": [".py"],
"md": [".md"],
"txt": [".txt"],
"json": [".json"],
"all": [".py", ".md", ".txt", ".json", ".yaml", ".yml"]
}.get(file_type, [".py", ".md", ".txt", ".json"])
results = []
pattern = re.compile(keyword, re.IGNORECASE)
for ext in extensions:
for file_path in dir_path.rglob(f"*{ext}"):
try:
with open(file_path, "r", encoding="utf-8", errors="ignore") as f:
for line_no, line in enumerate(f, 1):
if pattern.search(line):
results.append({
"file": str(file_path),
"line": line_no,
"content": line.strip()[:200]
})
if len(results) >= max_results:
break
except Exception:
continue
if len(results) >= max_results:
break
if len(results) >= max_results:
break
self.search_history.append({
"keyword": keyword,
"directory": directory,
"results_count": len(results)
})
return [types.TextContent(
text=json.dumps({
"keyword": keyword,
"total_matches": len(results),
"results": results
}, ensure_ascii=False, indent=2)
)]
async def _handle_file_read(
self,
path: str,
start_line: int = 1,
end_line: int = 100
) -> list[types.ContentBlock]:
"""读取文件指定行范围"""
file_path = Path(path)
if not file_path.exists():
return [types.TextContent(
text=json.dumps({"error": f"文件不存在: {path}"}, ensure_ascii=False)
)]
try:
with open(file_path, "r", encoding="utf-8", errors="ignore") as f:
lines = f.readlines()
# 提取指定范围（处理越界情况）
start = max(0, start_line - 1)
end = min(len(lines), end_line)
content_lines = lines[start:end]
result = {
"path": str(file_path),
"total_lines": len(lines),
"read_range": f"{start_line}-{end_line}",
"content": "".join(content_lines)
}
return [types.TextContent(
text=json.dumps(result, ensure_ascii=False, indent=2)
)]
except Exception as e:
return [types.TextContent(
text=json.dumps({"error": str(e)}, ensure_ascii=False)
)]
# ========== MCP Server启动 ==========
async def main():
"""MCP Server主入口"""
server = FileSearchServer()
# 使用stdio传输层启动
# stdin接收请求，stdout返回响应
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
await mcp.server.Server(
name="file-search-server",
version="1.0.0",
tools=server.get_tools(),
).run(
read_stream,
write_stream,
server.handle_tool_call,
)
if __name__ == "__main__":
asyncio.run(main())
4.3 调试与验证
# 方式1：使用MCP Inspector调试
npx @anthropic-ai/mcp-inspector python server.py
# 方式2：直接运行测试stdio通信
echo '{"jsonrpc":"2.0","id":1,"method":"tools/list","params":{}}' \
| python server.py
# 预期输出：返回tools列表（JSON格式）
4.4 接入Claude Desktop
// ~/.config/Claude/claude_desktop_config.json
{
"mcpServers": {
"file-search": {
"command": "python",
"args": ["/path/to/server.py"]
}
}
}
重启Claude Desktop后，说出指令：
“帮我搜索
/project/src目录下所有包含async def的Python文件”
Claude会通过MCP协议自动调用 file_search 工具。
5. 生产级实战：构建RAG检索MCP Server
5.1 需求场景
构建一个向量检索MCP Server，让AI能够：
- 检索知识库中的相关文档
- 获取文档片段作为上下文
- 支持语义相似度搜索
5.2 完整实现
#!/usr/bin/env python3
"""
RAG检索MCP Server
功能：提供向量语义检索能力，支持本地知识库问答
依赖：pip install mcp chromadb sentence-transformers
"""
import mcp.server.stdio
import mcp.types as types
import chromadb
from chromadb.config import Settings
from sentence_transformers import SentenceTransformer
import json
import asyncio
import os
from pathlib import Path
class RAGMCPServer:
"""基于向量数据库的RAG检索服务"""
def __init__(
self,
collection_name: str = "knowledge_base",
model_name: str = "paraphrase-multilingual-MiniLM-L12-v2"
):
# 初始化向量数据库（Chroma持久化存储）
self.client = chromadb.Client(Settings(
persist_directory="./chroma_db",
anonymized_telemetry=False
))
# 获取或创建集合
try:
self.collection = self.client.get_collection(collection_name)
except Exception:
self.collection = self.client.create_collection(
name=collection_name,
metadata={"description": "知识库向量集合"}
)
# 加载Embedding模型
self.encoder = SentenceTransformer(model_name)
self.collection_name = collection_name
# 索引映射（ID → 元数据）
self._id_metadata: dict[str, dict] = {}
def get_tools(self) -> list[types.Tool]:
return [
types.Tool(
name="rag_search",
description=(
"在知识库中检索与查询最相关的文档片段。"
"使用语义相似度搜索，返回最相关的N条结果。"
"适用于：技术问题解答、文档查询、代码检索等场景。"
),
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "语义检索查询"
},
"top_k": {
"type": "integer",
"description": "返回的最相关结果数量",
"default": 5,
"minimum": 1,
"maximum": 20
},
"min_similarity": {
"type": "number",
"description": "最小相似度阈值（0-1）",
"default": 0.5,
"minimum": 0.0,
"maximum": 1.0
}
},
"required": ["query"]
}
),
types.Tool(
name="rag_index_document",
description=(
"将文档内容索引到知识库。"
"支持自动分块、向量化和存储。"
"块大小默认500字符，重叠50字符。"
),
inputSchema={
"type": "object",
"properties": {
"content": {
"type": "string",
"description": "要索引的文档内容"
},
"doc_id": {
"type": "string",
"description": "文档唯一标识"
},
"metadata": {
"type": "object",
"description": "文档元数据（如标题、来源、时间）",
"properties": {
"title": {"type": "string"},
"source": {"type": "string"},
"tags": {"type": "array", "items": {"type": "string"}}
}
},
"chunk_size": {
"type": "integer",
"description": "分块大小（字符数）",
"default": 500,
"minimum": 100,
"maximum": 2000
}
},
"required": ["content", "doc_id"]
}
),
types.Tool(
name="rag_get_stats",
description="获取知识库统计信息：文档数、块数、集合配置",
inputSchema={
"type": "object",
"properties": {}
}
)
]
async def handle_tool_call(
self,
name: str,
arguments: dict
) -> list[types.ContentBlock]:
if name == "rag_search":
return await self._handle_search(**arguments)
elif name == "rag_index_document":
return await self._handle_index(**arguments)
elif name == "rag_get_stats":
return await self._handle_stats()
raise ValueError(f"Unknown tool: {name}")
def _chunk_text(self, text: str, chunk_size: int, overlap: int = 50) -> list[str]:
"""将长文本按指定大小分块，支持重叠"""
chunks = []
start = 0
while start < len(text):
end = start + chunk_size
chunks.append(text[start:end])
start = end - overlap
return chunks
async def _handle_index(
self,
content: str,
doc_id: str,
metadata: dict = None,
chunk_size: int = 500
) -> list[types.ContentBlock]:
"""索引文档"""
metadata = metadata or {}
# 分块处理
chunks = self._chunk_text(content, chunk_size)
# 批量生成向量
embeddings = self.encoder.encode(chunks).tolist()
# 批量添加（带ID）
ids = [f"{doc_id}_chunk_{i}" for i in range(len(chunks))]
# 存储元数据
for i, chunk in enumerate(chunks):
self._id_metadata[ids[i]] = {
"doc_id": doc_id,
"chunk_index": i,
"total_chunks": len(chunks),
**metadata
}
self.collection.add(
embeddings=embeddings,
documents=chunks,
ids=ids,
metadatas=[self._id_metadata[uid] for uid in ids]
)
return [types.TextContent(
text=json.dumps({
"success": True,
"doc_id": doc_id,
"chunks_indexed": len(chunks),
"chunk_size": chunk_size
}, ensure_ascii=False)
)]
async def _handle_search(
self,
query: str,
top_k: int = 5,
min_similarity: float = 0.5
) -> list[types.ContentBlock]:
"""语义检索"""
# 生成查询向量
query_embedding = self.encoder.encode([query]).tolist()[0]
# 向量检索
results = self.collection.query(
query_embeddings=[query_embedding],
n_results=top_k,
include=["documents", "metadatas", "distances"]
)
# 整理结果
documents = results.get("documents", [[]])[0]
metadatas = results.get("metadatas", [[]])[0]
distances = results.get("distances", [[]])[0]
# 过滤低相似度结果（Chroma用L2距离，转为相似度）
similarity_scores = [1 - d / 2 for d in distances] # 近似转换
filtered = [
{
"content": doc,
"metadata": meta,
"similarity": round(score, 4)
}
for doc, meta, score in zip(documents, metadatas, similarity_scores)
if score >= min_similarity
]
return [types.TextContent(
text=json.dumps({
"query": query,
"results_count": len(filtered),
"results": filtered
}, ensure_ascii=False, indent=2)
)]
async def _handle_stats(self) -> list[types.ContentBlock]:
return [types.TextContent(
text=json.dumps({
"collection_name": self.collection_name,
"total_chunks": self.collection.count(),
"unique_docs": len(set(
m.get("doc_id")
for m in self._id_metadata.values()
)),
"embedding_model": self.encoder.model_name
}, ensure_ascii=False)
)]
async def main():
server = RAGMCPServer()
async with mcp.server.stdio.stdio_server() as (read, write):
await mcp.server.Server(
name="rag-search-server",
version="1.0.0",
tools=server.get_tools()
).run(read, write, server.handle_tool_call)
if __name__ == "__main__":
asyncio.run(main())
5.3 完整使用流程
# 使用示例：通过Claude自然语言调用
"""
用户: "我之前写过一篇关于PostgreSQL索引优化的笔记，
里面提到BRIN索引的适用场景，能帮我找出来吗？"
Claude通过MCP调用:
→ rag_search(query="PostgreSQL BRIN索引 适用场景", top_k=5)
返回结果（JSON格式）:
{
"query": "PostgreSQL BRIN索引 适用场景",
"results_count": 2,
"results": [
{
"content": "BRIN索引适合时序数据...\n列如日志表...",
"metadata": {"doc_id": "postgres_notes_001", "title": "索引优化实践"},
"similarity": 0.8723
}
]
}
"""
6. 当前局限性与未来方向
6.1 当前版本的局限性
| 问题 | 说明 | 当前解决方案 |
|---|---|---|
| 无状态限制 | stdio模式下无会话管理，复杂Agent需要自己维护状态 | 外部状态存储（如Redis） |
| 安全保障缺失 | Server可以读写任意文件/执行任意命令 | 目前靠信任链，未来需要权限模型 |
| 性能瓶颈 | Python GIL限制高并发场景 | 可用uvicorn包装为HTTP服务 |
| 模型适配 | 各AI模型对MCP的Tool调用策略不同 | 需要针对模型调整Prompt |
| 调试困难 | stdio模式下错误信息不够友好 | 使用MCP Inspector辅助调试 |
6.2 社区动态与未来方向
根据2025-2026年MCP社区的演进趋势：
2025 Q1: 工具发现协议稳定
2025 Q2: MCP Hub发布，支持Server注册与发现
2025 Q3: 安全认证模型草案（MCP Auth）
2025 Q4: 多模态MCP扩展（图像、音频工具支持）
2026 Q1: 企业级MCP Gateway（认证、限流、审计）
7. 总结与参考资料
7.1 核心要点回顾
┌──────────────────────────────────────────────────────────┐
│ MCP 核心知识图谱 │
├──────────────────────────────────────────────────────────┤
│ 协议架构：Application Layer → Protocol Layer → Transport │
│ 传输方式：stdio（本地）| HTTP+SSE（远程） │
│ 消息格式：JSON-RPC 2.0 │
│ 三大能力：Tools（工具调用）/ Resources（资源访问）/ │
│ Prompts（提示模板） │
│ 参数验证：JSON Schema + 类型检查 │
│ 工具发现：tools/list → tools/call 两步流程 │
└──────────────────────────────────────────────────────────┘
7.2 快速开发 Checklist
- 明确Server能力边界（不要做太多，也不要太少）
- 工具description写清楚（这是AI理解你的工具的唯一依据）
- inputSchema用JSON Schema规范定义
- 参数验证要全面（防止AI传非法参数）
- 返回结果用结构化JSON，便于AI解析
- 添加错误处理，不要让Server直接crash
- 优先实现stdio模式，验证通过后再扩展HTTP
7.3 参考资料
写在最后
MCP的价值不在于「有没有」，而在于「生态成熟度」。当前阶段，它解决的是协议层的问题，但工具的安全性、性能、企业级治理，还需要社区进一步推进。如果你有具体的MCP开发场景，欢迎在评论区交流。
相关项目代码已上传至：[GitHub仓库链接]（如需）
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
7
0- 0
已为社区贡献1条内容
所有评论(0)