filmorago pro会员解锁专业版
84.37M · 2025-10-04
Model Context Protocol (MCP) 是 Anthropic 推出的开源标准协议，用于连接 AI 应用与外部系统。如果用一个形象的比喻，MCP 就像是 AI 应用的 USB-C 接口——提供了一个标准化的连接方式，让 AI 应用能够与各种数据源、工具和工作流无缝集成。
当前 AI 应用面临的核心挑战：
MCP 通过提供统一的协议标准，将这些碎片化的集成方案替换为单一的标准协议，从根本上解决了这些问题。
MCP 采用经典的客户端-服务器架构，包含以下核心组件：
┌─────────────────┐
│ MCP Host │ ← AI 应用（如 Claude Desktop）
│ (协调者) │
└────────┬────────┘
│
├─── MCP Client 1 ──→ MCP Server A (数据库)
│
├─── MCP Client 2 ──→ MCP Server B (文件系统)
│
└─── MCP Client 3 ──→ MCP Server C (API 服务)
各组件职责：
MCP 基于 JSON-RPC 2.0 构建，包含两个核心层：
MCP 是一个有状态协议，通信流程如下：
sequenceDiagram
Client->>Server: Initialize Request(客户端能力声明)
Server-->>Client: Initialize Response(服务器能力声明)
Client-)Server: List Tools Request
Server-->>Client:Tools List
Client-)Server: Tool Call Request
Server-->>Client: Tool Result
Server-->>Client:Notification (optional)(资源变更通知)
关键特性：
MCP 服务器可以向客户端提供三种类型的能力：
类似文件系统的只读数据源，供客户端读取。
典型应用场景：
示例：
{
"uri": "file:///path/to/document.txt",
"name": "项目文档",
"mimeType": "text/plain",
"description": "项目需求文档"
}
LLM 可调用的可执行函数，实现具体操作。
典型应用场景：
Python 实现示例：
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("weather")
@mcp.tool()
async def get_weather(city: str, state: str) -> str:
"""获取指定城市的天气信息"""
# 实现天气查询逻辑
return f"{city}, {state} 的天气是晴天，温度 25°C"
预编写的任务模板，用于标准化常见操作。
典型应用场景：
示例：
{
"name": "code_review",
"description": "代码审查提示模板",
"arguments": [
{
"name": "language",
"description": "编程语言",
"required": true
}
]
}
让我们通过一个实际示例来理解如何构建 MCP 服务器。
from mcp.server.fastmcp import FastMCP
import sqlite3
# 初始化服务器
mcp = FastMCP("database-server")
@mcp.tool()
async def query_users(limit: int = 10) -> str:
"""查询用户列表"""
conn = sqlite3.connect('app.db')
cursor = conn.cursor()
cursor.execute('SELECT * FROM users LIMIT ?', (limit,))
results = cursor.fetchall()
conn.close()
return str(results)
@mcp.tool()
async def create_user(name: str, email: str) -> str:
"""创建新用户"""
conn = sqlite3.connect('app.db')
cursor = conn.cursor()
cursor.execute(
'INSERT INTO users (name, email) VALUES (?, ?)',
(name, email)
)
conn.commit()
user_id = cursor.lastrowid
conn.close()
return f"用户创建成功，ID: {user_id}"
# 启动服务器（使用 stdio 传输）
if __name__ == "__main__":
mcp.run()
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
const server = new Server(
{
name: "file-server",
version: "1.0.0",
},
{
capabilities: {
tools: {},
},
}
);
// 注册工具
server.setRequestHandler("tools/list", async () => {
return {
tools: [
{
name: "read_file",
description: "读取文件内容",
inputSchema: {
type: "object",
properties: {
path: {
type: "string",
description: "文件路径",
},
},
required: ["path"],
},
},
],
};
});
// 处理工具调用
server.setRequestHandler("tools/call", async (request) => {
if (request.params.name === "read_file") {
const fs = await import("fs/promises");
const content = await fs.readFile(request.params.arguments.path, "utf-8");
return {
content: [
{
type: "text",
text: content,
},
],
};
}
throw new Error("未知工具");
});
// 启动服务器
const transport = new StdioServerTransport();
await server.connect(transport);
from anthropic import Anthropic
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
class MCPClient:
def __init__(self):
self.anthropic = Anthropic()
self.session = None
async def connect_to_server(self, server_script_path: str):
"""连接到 MCP 服务器"""
server_params = StdioServerParameters(
command="python",
args=[server_script_path],
)
stdio_transport = await stdio_client(server_params)
self.stdio, self.write = stdio_transport
self.session = ClientSession(self.stdio, self.write)
await self.session.initialize()
async def process_query(self, query: str) -> str:
"""处理用户查询"""
# 获取可用工具
tools = await self.session.list_tools()
# 发送查询到 AI 模型
response = self.anthropic.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=4096,
messages=[{"role": "user", "content": query}],
tools=[{
"name": tool.name,
"description": tool.description,
"input_schema": tool.inputSchema
} for tool in tools.tools]
)
# 处理工具调用
if response.stop_reason == "tool_use":
tool_results = []
for content in response.content:
if content.type == "tool_use":
result = await self.session.call_tool(
content.name,
content.input
)
tool_results.append({
"type": "tool_result",
"tool_use_id": content.id,
"content": result.content
})
# 将工具结果发送回 AI
final_response = self.anthropic.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=4096,
messages=[
{"role": "user", "content": query},
{"role": "assistant", "content": response.content},
{"role": "user", "content": tool_results}
]
)
return final_response.content[0].text
return response.content[0].text
async def main():
client = MCPClient()
# 连接到数据库服务器
await client.connect_to_server("database_server.py")
# 处理查询
result = await client.process_query(
"帮我创建一个名为张三，邮箱是[email protected]的用户"
)
print(result)
# 运行
import asyncio
asyncio.run(main())
Anthropic 提供了多个参考实现：
| 服务器 | 功能描述 |
|---|---|
| filesystem | 安全的文件系统操作，支持访问控制 |
| git | Git 仓库读取、搜索和操作 |
| github | GitHub API 集成，仓库和问题管理 |
| fetch | Web 内容抓取和转换 |
| memory | 基于知识图谱的持久化记忆系统 |
| sequential-thinking | 动态问题解决的思维链 |
已集成 MCP 的公司和产品：
| 维度 | 传统方式 | MCP 方式 |
|---|---|---|
| 开发成本 | 每个数据源需单独实现 | 一次实现，到处复用 |
| 维护难度 | 多套代码分别维护 | 统一协议标准 |
| 扩展性 | 添加新数据源需重写集成 | 实现 MCP 服务器即可 |
| 互操作性 | 厂商方案互不兼容 | 开放标准，跨平台 |
| 上下文保持 | 工具切换丢失上下文 | 协议层保持状态 |
MCP 不是替代 Function Calling，而是更高层次的抽象：
MCP 实际上是通过 Function Calling 来调用工具，但提供了：
// 客户端请求
{
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {
"roots": {
"listChanged": true
},
"sampling": {}
},
"clientInfo": {
"name": "ExampleClient",
"version": "1.0.0"
}
}
}
// 服务器响应
{
"jsonrpc": "2.0",
"id": 1,
"result": {
"protocolVersion": "2024-11-05",
"capabilities": {
"logging": {},
"prompts": {
"listChanged": true
},
"resources": {
"subscribe": true,
"listChanged": true
},
"tools": {
"listChanged": true
}
},
"serverInfo": {
"name": "ExampleServer",
"version": "1.0.0"
}
}
}
MCP 使用能力声明来确定客户端和服务器支持的功能：
客户端可声明的能力：
roots：支持项目根目录管理sampling：支持 LLM 采样请求服务器可声明的能力：
tools：提供可调用工具resources：提供数据资源prompts：提供提示模板logging：支持日志记录服务器可以主动向客户端推送变更通知：
{
"jsonrpc": "2.0",
"method": "notifications/resources/list_changed",
"params": {}
}
客户端收到通知后，可以重新获取资源列表以获得最新状态。
需求：构建一个能访问用户日历、邮件和任务管理系统的 AI 助手
实现方案：
# calendar_server.py
@mcp.tool()
async def get_today_events() -> str:
"""获取今天的日程安排"""
# 连接 Google Calendar API
events = calendar_api.get_events(date=today)
return format_events(events)
@mcp.tool()
async def create_event(title: str, start_time: str, duration: int) -> str:
"""创建新的日程事件"""
event = calendar_api.create_event(title, start_time, duration)
return f"事件创建成功: {event.id}"
用户可以自然语言交互：
需求：AI 根据设计规范自动生成代码
实现方案：
# design_spec_server.py
@mcp.resource()
async def get_design_spec(component_name: str) -> str:
"""获取组件设计规范"""
spec = load_spec(component_name)
return spec.to_json()
@mcp.tool()
async def generate_component(spec: dict) -> str:
"""根据规范生成组件代码"""
code = code_generator.generate(spec)
return code
需求：跨部门数据的统一分析
实现方案：
# Multi-server setup
servers = [
"sales_db_server", # 销售数据库
"hr_api_server", # 人力资源 API
"finance_server", # 财务系统
]
# AI 可以跨系统查询
query = """
分析一下上季度销售业绩与人员投入的关系，
并给出下季度的人力资源建议
"""
访问控制
@mcp.tool()
async def read_file(path: str) -> str:
# 验证路径在允许的目录内
if not is_path_allowed(path):
raise PermissionError("路径访问被拒绝")
return read_file_content(path)
参数验证
from pydantic import BaseModel, validator
class QueryParams(BaseModel):
limit: int
@validator('limit')
def validate_limit(cls, v):
if v > 1000:
raise ValueError('limit 不能超过 1000')
return v
敏感信息处理
连接池管理
from contextlib import asynccontextmanager
class DatabaseServer:
def __init__(self):
self.pool = create_connection_pool()
@asynccontextmanager
async def get_connection(self):
conn = await self.pool.acquire()
try:
yield conn
finally:
await self.pool.release(conn)
缓存策略
from functools import lru_cache
@mcp.tool()
@lru_cache(maxsize=100)
async def get_static_data(key: str) -> str:
return expensive_computation(key)
并发控制
import asyncio
semaphore = asyncio.Semaphore(10) # 限制并发数
@mcp.tool()
async def api_call(endpoint: str) -> str:
async with semaphore:
return await fetch(endpoint)
更丰富的传输层支持
智能能力发现
企业级特性
MCP 正在构建一个协作式的开源生态：
[[mcp_servers]] 片段定义要挂载的 Server（含 command、args、env、工作目录、
速率限制）。它默认把 Server 包在 Codex 的沙箱与审批体系里：每个工具调用都经
过权限检查，长命令会触发“approval”交互，同时把 Server 输出写入结构化日志，供
codex.log 与审计面板消费。底层整合了本地文件树索引、诊断（lint/test）和 plan/
review 工作流，并以 MCP 资源流的形式把这些上下文塞给模型。优点是安全治理彻底、
与 CLI 体验绑定紧密；缺点是可视化和多窗口体验有限，Server 生命周期基本受 Codex
控制，外部想复用需额外做桥接。
配置文件 claude_desktop_config.json 或 VS Code 插件设置里的 mcpServers 字段声
明 MCP Server。它把文件树、终端、Run&Debug 面板对齐到同个会话里，Claude 输出时
会流式显示 modelOutput、toolProgress，并允许用户在工具调用前直接修改参数或拿到
diff。Anthropic 的参考实现大量使用 TS SDK，支持热加载 Server、per-project 覆
盖、以及 UI 级别的冲突提示。Claude Code 的强项是 IDE 深度集成与 Prompt 工程资
产管理；不足在于 CLI 自动化较弱，而且对权限/审批的内置治理比 Codex 稍轻，更多
依赖 IDE 环境或企业版的外围策略。
实现，在 genim.config.ts 里用类 Graph 方式声明 MCP Server、资源映射与触发
器。Genim 会把每次模型交互拆成显式的 Node：ResourceFetch、PromptTemplate、
ToolCall，并在 CLI 中输出 DAG 式追踪，方便把上下文拼装过程纳入 CI/CD。它默认支
持把 Server 运行在 Docker/Nix profile 里，强调 reproducible run，以及 CLI 中的
genim run、genim diff。长处是自动化友好、易嵌入现有流水线；短板是交互式体验偏
弱，安全策略需要用户自己在 Docker profile 或 Server 里兜底。
- 治理与安全：Codex CLI > Genim CLI > Claude Code。Codex 的审批流和沙箱最完
整，Genim 提供容器化但策略自管，Claude Code 更依赖 IDE 权限。
- 交互体验：Claude Code 在 IDE 内的 UI、上下文回放最强；Codex 在高级 CLI 操作
与调试方面领先；Genim 着重自动化，没有图形界面。
- 生态与可扩展性：Genim 把工作流开放给脚本最方便；Codex/Claude 都能挂任意 MCP
Server，但前者耦合 Codex 沙箱，后者耦合 IDE 环境。
- 默认能力：Codex 自带大量内置 Server（shell、fs、git、issue）；Claude 依赖
IDE 插件生态；Genim 倚重用户自定义。
谁更适合哪类团队****
- 想要“安全高可信、命令行驱动”的团队：Codex CLI。
- 需要“IDE 即时回馈、开发体验优先”的团队：Claude Code。
- 希望“把上下文拼装纳入自动化流水线”的团队：Genim CLI。
- 上下文编排层标准化：三者都在尝试把 RAG、工具、Prompt 合成流程显式化，下一步
会朝“可视化 DAG + 版本化上下文包”演进。
- 细粒度权限治理：未来 MCP Host 预计会原生支持资源/工具级 RBAC、Secret 委托、
审计事件流。Codex 走得最前，但还需跨 Host 对齐。
- 多代理协作：把多个 MCP Host/Server 串联，让不同能力代理共享上下文，这对
Genim 的流水线和 Codex 的 plan/review 模式是自然进化。
- 端到端可观测性：谁能把 modelOutput、toolProgress、resourceUpdated 串成统一追
踪，谁就更易进企业。Genim 已有 DAG 可视化雏形，值得关注。
如需继续深挖，建议：1) 实测在 Codex/Claude/Genim 下挂同一个 Server（如 mcp-
git），比较上下文注入效果；2) 根据自身安全需求选主力 Host，再看是否要做二次封
装或把三个结合使用。
MCP (Model Context Protocol) 代表了 AI 应用集成的范式转变：
MCP 不仅仅是一个技术协议，更是AI 应用走向标准化、工程化的重要一步。随着生态的发展，它有潜力成为 AI 应用集成的事实标准。
参考资源
关于作者
如果你对 AI 应用开发和工程化实践感兴趣，欢迎关注我的掘金账号，一起探讨技术！
84.37M · 2025-10-04
97.24MB · 2025-10-04
132.36MB · 2025-10-04
