- 引言
- MCP 架构概览
- MCP 核心原语（Primitives）
- 用 Python FastMCP 实现 MCP 服务器
- 用 TypeScript SDK 实现 MCP 服务器
- stdio 与 SSE 两种传输方式
- Claude Desktop 集成
- 实现 MCP 客户端
- 智能体系统：LangGraph + MCP
- 需要用户批准的危险工具
- 安全考量
- MCP 与传统 function calling 的比较
- 小测验
- 结语
引言
AI 要从单纯的文本生成器，进化为真正使用工具的智能体，就需要模型与外部系统之间一套标准化的通信方式。Anthropic 于 2024 年 11 月发布的 Model Context Protocol（MCP），正是解决这个问题的开放标准。
正如 USB-C 用一套标准连接各种设备，MCP 让 LLM 能以统一的方式与任何数据源、任何工具通信。本指南将完整覆盖 MCP 的架构、实战服务器实现、Claude Desktop 集成，直到智能体系统的搭建。
MCP 架构概览
基于 JSON-RPC 2.0 的通信
MCP 运行在 JSON-RPC 2.0 协议之上。客户端发送请求、服务器返回响应，结构看似简单，但其上叠加了一层强大的抽象。
{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": "read_file",
"arguments": {
"path": "/tmp/example.txt"
}
}
}
Hosts / Clients / Servers 三角结构
MCP 由三种角色构成：
- Host：运行 LLM 的应用程序，例如 Claude Desktop、VS Code 或自定义 AI 应用
- Client：Host 内部负责与某个 MCP 服务器管理 1:1 会话的组件
- Server：向外部提供实际工具、资源、提示词的轻量级进程
[Host: Claude Desktop]
└── [Client] ──── stdio/SSE ──── [MCP Server: filesystem]
└── [Client] ──── stdio/SSE ──── [MCP Server: github]
└── [Client] ──── stdio/SSE ──── [MCP Server: database]
Host 可以同时连接多个 MCP 服务器，并把各服务器提供的工具暴露给 LLM。
MCP 核心原语（Primitives）
Resources — 只读数据
Resources 是服务器向客户端暴露的只读数据。文件、数据库记录、API 响应等，都通过标准化的 URI 访问。
- URI 格式：
file:///home/user/docs/report.pdf、db://mydb/users/42 - 没有副作用（side effect）——纯粹的数据读取
- 支持文本或二进制内容
Tools — 函数执行
Tools 是 LLM 可以执行的函数。与 Resources 不同，Tools 可以带有副作用。
- 创建/修改文件、调用 API、写数据库等
- 用 JSON Schema 定义输入参数
- 执行结果以文本或图像形式返回
Prompts — 可复用模板
Prompts 是服务器提供的提示词模板。它把用户反复使用的模式在服务器一侧标准化。
@mcp.prompt()
def code_review_prompt(language: str, code: str) -> str:
return f"请审查这段 {language} 代码并给出改进建议：\n\n{code}"
Sampling — 由服务器发起的 LLM 调用
Sampling 是 MCP 的一项独特功能，它允许服务器通过客户端反向调用 LLM。适用于服务器在复杂处理过程中需要 AI 判断的场景。
用 Python FastMCP 实现 MCP 服务器
安装与基本结构
pip install fastmcp
FastMCP 提供基于装饰器的简洁 API，可以快速构建 MCP 服务器。
from fastmcp import FastMCP
mcp = FastMCP("filesystem-server")
@mcp.tool()
def read_file(path: str) -> str:
"""读取并返回文件内容。"""
with open(path, "r", encoding="utf-8") as f:
return f.read()
@mcp.tool()
def write_file(path: str, content: str) -> str:
"""向文件写入内容。"""
with open(path, "w", encoding="utf-8") as f:
f.write(content)
return f"文件保存完成：{path}"
@mcp.resource("file://{path}")
def get_file_resource(path: str) -> str:
"""通过 URI 提供文件资源。"""
with open(path, "r", encoding="utf-8") as f:
return f.read()
if __name__ == "__main__":
mcp.run()
添加目录浏览工具
import os
from pathlib import Path
from fastmcp import FastMCP
mcp = FastMCP("filesystem-server")
@mcp.tool()
def list_directory(path: str = ".") -> list[str]:
"""返回目录中的文件列表。"""
p = Path(path)
if not p.is_dir():
raise ValueError(f"{path} 不是一个目录")
return [str(item) for item in p.iterdir()]
@mcp.tool()
def search_files(directory: str, pattern: str) -> list[str]:
"""在目录中搜索匹配指定模式的文件。"""
p = Path(directory)
return [str(f) for f in p.rglob(pattern)]
if __name__ == "__main__":
mcp.run(transport="stdio")
用 TypeScript SDK 实现 MCP 服务器
用 TypeScript 实现一个提供数据库资源的 MCP 服务器。
import { McpServer, ResourceTemplate } from '@modelcontextprotocol/sdk/server/mcp.js'
import { StdioServerTransport } from '@modelcontextprotocol/sdk/server/stdio.js'
import { z } from 'zod'
const server = new McpServer({
name: 'database-server',
version: '1.0.0',
})
// 资源：把数据库记录通过 URI 暴露出来
server.resource(
'user',
new ResourceTemplate('db://users/{id}', { list: undefined }),
async (uri, { id }) => {
const user = await fetchUserById(id)
return {
contents: [
{
uri: uri.href,
text: JSON.stringify(user, null, 2),
mimeType: 'application/json',
},
],
}
}
)
// 工具：执行 SQL 查询
server.tool('query_db', { sql: z.string().describe('要执行的 SQL 查询') }, async ({ sql }) => {
const results = await executeQuery(sql)
return {
content: [
{
type: 'text',
text: JSON.stringify(results, null, 2),
},
],
}
})
async function main() {
const transport = new StdioServerTransport()
await server.connect(transport)
}
main()
stdio 与 SSE 两种传输方式
stdio 传输
适合同一台机器上的进程间通信。当 Claude Desktop 把服务器当作子进程启动时使用。
- 优点：设置简单、无需额外网络、安全（仅限本地）
- 缺点：无法远程访问、只支持单个客户端
- 使用场景：Claude Desktop、本地开发环境、CI 流水线
SSE（Server-Sent Events）传输
适合基于 HTTP 的远程通信。当多个客户端要连接同一个服务器时使用。
- 优点：可远程访问、支持多客户端、便于 Web 集成
- 缺点：需要搭建 HTTP 服务器、需要实现认证
- 使用场景：团队共享的 MCP 服务器、云端部署、多用户环境
# SSE 服务器启动示例
if __name__ == "__main__":
mcp.run(transport="sse", host="0.0.0.0", port=8000)
Claude Desktop 集成
配置 claude_desktop_config.json
在 macOS 上，编辑 ~/Library/Application Support/Claude/claude_desktop_config.json。
{
"mcpServers": {
"filesystem": {
"command": "python",
"args": ["/path/to/filesystem_server.py"],
"env": {
"ALLOWED_DIRS": "/home/user/documents,/tmp"
}
},
"github": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-github"],
"env": {
"GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_your_token_here"
}
},
"sqlite": {
"command": "uvx",
"args": ["mcp-server-sqlite", "--db-path", "/path/to/database.db"]
}
}
}
确认本地服务器已生效
重启 Claude Desktop 后，已连接服务器的工具会自动可用。可以在聊天窗口下方的工具图标里查看可用的 MCP 工具列表。
实现 MCP 客户端
除了服务器，你也可以自己实现 MCP 客户端来连接服务器。
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
async def main():
server_params = StdioServerParameters(
command="python",
args=["filesystem_server.py"],
)
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
# 初始化
await session.initialize()
# 获取可用工具列表
tools = await session.list_tools()
print("工具列表：", [t.name for t in tools.tools])
# 调用工具
result = await session.call_tool(
"read_file",
arguments={"path": "/tmp/test.txt"}
)
print("结果：", result.content[0].text)
asyncio.run(main())
智能体系统：LangGraph + MCP
把 LangGraph 与 MCP 连接起来，就能让 LangGraph 智能体直接使用 MCP 服务器提供的工具。
from langchain_mcp_adapters.tools import load_mcp_tools
from langgraph.prebuilt import create_react_agent
from langchain_anthropic import ChatAnthropic
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
import asyncio
async def create_mcp_agent():
model = ChatAnthropic(model="claude-3-5-sonnet-20241022")
server_params = StdioServerParameters(
command="python",
args=["filesystem_server.py"],
)
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
# 把 MCP 工具转换为 LangChain 工具
tools = await load_mcp_tools(session)
# 创建 ReAct 智能体
agent = create_react_agent(model, tools)
result = await agent.ainvoke({
"messages": [{"role": "user", "content": "请列出 /tmp 目录下的文件"}]
})
return result
asyncio.run(create_mcp_agent())
需要用户批准的危险工具
对敏感操作，实现一套请求用户确认的模式。
from fastmcp import FastMCP
from fastmcp.exceptions import ToolError
mcp = FastMCP("safe-server")
DANGEROUS_PATHS = ["/etc", "/sys", "/proc", "/root"]
@mcp.tool()
def delete_file(path: str, confirmed: bool = False) -> str:
"""
删除文件。删除前必须显式设置 confirmed=True。
Args:
path: 要删除的文件路径
confirmed: 显式确认删除意图（默认值：False）
"""
import os
from pathlib import Path
abs_path = str(Path(path).resolve())
# 阻止危险路径
for danger in DANGEROUS_PATHS:
if abs_path.startswith(danger):
raise ToolError(f"无法删除 {danger} 下的路径")
# 未经确认调用时，向用户请求确认
if not confirmed:
return (
f"警告：即将永久删除 {abs_path}。"
f"请设置 confirmed=True 后再次调用以完成删除。"
)
os.remove(abs_path)
return f"删除完成：{abs_path}"
安全考量
权限模型
MCP 服务器应遵循最小权限原则。
- 只允许访问必要的目录、数据库
- 用环境变量管理允许访问的路径列表
- 防御路径穿越（Path traversal）攻击
import os
from pathlib import Path
ALLOWED_DIRS = os.getenv("ALLOWED_DIRS", "/tmp").split(",")
def is_path_allowed(path: str) -> bool:
abs_path = Path(path).resolve()
return any(
str(abs_path).startswith(allowed.strip())
for allowed in ALLOWED_DIRS
)
敏感数据处理
- API 密钥、密码通过环境变量注入，禁止硬编码进代码
- 禁止在日志中记录敏感的响应内容
- 禁止在没有 TLS 的情况下运行远程 MCP 服务器
沙箱化
在生产环境中，应把 MCP 服务器隔离在容器或虚拟环境中运行。
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY server.py .
# 使用专用的非 root 用户运行
RUN useradd -m mcpuser
USER mcpuser
CMD ["python", "server.py"]
Tool Call 校验
LLM 生成的 tool call 参数，服务器一侧必须重新校验。LLM 可能被提示词注入攻击，从而注入恶意参数。
from pydantic import BaseModel, validator
class FileReadParams(BaseModel):
path: str
@validator("path")
def validate_path(cls, v):
p = Path(v).resolve()
if ".." in str(p):
raise ValueError("检测到路径穿越尝试")
if not is_path_allowed(str(p)):
raise ValueError("路径不在允许列表中")
return str(p)
MCP 与传统 function calling 的比较
| 项目 | OpenAI Function Calling | MCP |
|---|---|---|
| 标准化 | 依赖特定厂商 | 开放标准 |
| 可复用性 | 每个模型都要重新实现 | 实现一次即可适配所有兼容模型 |
| 数据访问 | 无（仅工具） | 通过 Resources 直接暴露数据 |
| 反向 LLM 调用 | 无 | 通过 Sampling 让服务器调用 LLM |
| 传输方式 | 仅 HTTP | 支持 stdio、SSE |
小测验
Q1. MCP 中 Resources 与 Tools 的核心区别是什么？
答案：Resources 是只读的数据访问，没有副作用；Tools 是可以改变系统状态的函数执行。
解释：Resources 通过 file://、db:// 这类 URI 读取数据，不会改变系统状态。Tools 执行的是文件写入、API 调用、数据库写入等会改变状态的操作。LLM 需要读取数据时用 Resource，需要执行或改变某些内容时用 Tool。
Q2. 为什么说 MCP 比 OpenAI function calling 更有利于工具标准化？
答案：MCP 是厂商中立的开放标准，一份实现好的 MCP 服务器，可以在 Claude、GPT、开源模型等所有兼容模型上复用。
解释：OpenAI function calling 依赖于 OpenAI 的 API 格式，一旦切换模型，就得把工具全部重新实现。MCP 服务器在协议层面已经标准化，是与具体模型完全解耦的独立服务。此外，MCP 还提供了 function calling 所没有的能力，例如用 Resources 暴露数据、用 Prompts 共享模板、用 Sampling 实现反向 LLM 调用。
Q3. stdio 与 SSE 两种传输方式各自适合什么场景？
答案：stdio 适合本地单客户端环境（Claude Desktop、个人开发），SSE 适合远程、多客户端环境（团队共享服务器、云端部署）。
解释：stdio 由父进程直接启动子进程，不需要网络配置，安全性也更简单。SSE 会启动一个 HTTP 服务器，让多个客户端可以同时连接，因此更适合云端部署或团队内共享。不过使用 SSE 时，还需要额外配置认证和 TLS。
Q4. MCP Sampling 原语的工作方式与安全考量是什么？
答案：Sampling 是 MCP 服务器通过客户端（Host）反向调用 LLM 的功能，只有经过用户审查和批准之后才应该执行。
解释：通常的流程是 LLM 调用服务器的工具，而 Sampling 正好相反，是服务器向 LLM 请求"帮我分析这段文本"之类的操作。Host 需要在中间审查这类请求，只有在获得用户批准之后，才能把它转发给 LLM。必须防止恶意服务器利用 Sampling 无限制地调用 LLM，或者借此提取敏感内容。
Q5. 在 MCP 服务器中，针对危险工具的用户批准工作流该如何实现？
答案：使用 confirmed 参数模式——第一次调用返回警告，只有当用户在第二次调用中显式设置 confirmed=True 时，才真正执行操作。
解释：LLM 会读取 Tool 的描述（description）来决定如何调用它。当第一次调用（confirmed=False）返回"即将执行此操作，请以 confirmed=True 重新调用"这样明确的警告时，LLM 就会向用户请求确认。用户批准后，LLM 会以 confirmed=True 重新调用，真正的操作才会执行。这种模式应当与危险路径拦截、权限校验搭配使用。
结语
MCP 是 AI 智能体生态的 USB-C。标准只需定义一次，就能连接任意模型、任意客户端。得益于 Anthropic 把它作为开放标准发布，社区已经开发出数以百计的 MCP 服务器，VS Code、Cursor、JetBrains 等主流开发工具也开始支持 MCP。
用 Python FastMCP，5 分钟就能做出你的第一个服务器并接入 Claude Desktop。你会立刻感受到，让 AI 直接操作你的文件系统、数据库、API，是一种多么强大的体验。
현재 단락 (1/269)
AI 要从单纯的文本生成器，进化为**真正使用工具的智能体**，就需要模型与外部系统之间一套标准化的通信方式。Anthropic 于 2024 年 11 月发布的 **Model Context Pr...