MCP Server 提供什么
MCP Server 以标准化方式向 AI 应用暴露数据和能力。最常用的三个服务端原语是:
- Tools:执行计算、查询或产生外部操作;
- Resources:提供可以通过 URI 读取的数据;
- Prompts:提供可复用的消息模板。
它们并不只是 REST API 中 GET / POST 的简单替代。MCP 还定义了能力发现、输入 Schema、结构化内容、通知和错误语义,Host 可以据此决定向用户和模型提供哪些能力。
本文代码基于 MCP Python SDK v2。旧版 mcp.server.fastmcp.FastMCP 与 v2 的部分导入路径和 API 不兼容,迁移时应参考官方迁移指南。
1 2 3
| from mcp.server import MCPServer
mcp = MCPServer("Test Platform")
|
Tool 用于执行计算或访问外部系统。工具可以是只读查询,也可以产生副作用,因此“工具一定有副作用”并不准确。
一个工具定义至少应该明确:
- 名称和描述:帮助 Host 与模型理解何时使用;
- 输入类型:SDK 根据类型提示生成 JSON Schema;
- 输出类型:让调用方可以稳定解析结果;
- 错误语义:区分业务失败与系统异常;
- 权限和副作用:明确调用是否会读取敏感数据或改变外部状态。
基础工具
1 2 3 4 5 6 7 8 9
| from mcp.server import MCPServer
mcp = MCPServer("Math Server")
@mcp.tool() def add(a: int, b: int) -> int: """计算两个整数之和。""" return a + b
|
工具描述应该具体说明用途和边界,避免使用“万能工具”“可以处理所有请求”之类模糊描述。
结构化输出
优先使用 Pydantic、TypedDict、dataclass 或明确的字典类型返回结构化结果。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20
| from pydantic import BaseModel, Field from mcp.server import MCPServer
mcp = MCPServer("Weather Server")
class WeatherResult(BaseModel): city: str = Field(description="城市名称") temperature_c: float = Field(description="摄氏温度") condition: str = Field(description="天气情况")
@mcp.tool() def get_weather(city: str) -> WeatherResult: """查询指定城市的当前天气。""" return WeatherResult( city=city, temperature_c=22.0, condition="partly cloudy", )
|
结构化输出的好处包括:
- Client 可以根据
outputSchema 校验结果;
- Agent 不需要从自然语言中重新提取字段;
- 更适合后续工作流、测试和监控;
- 字段变化更容易进行版本管理。
错误处理
工具失败时,不要用伪造的成功结果掩盖异常。业务错误应返回可理解的信息,系统错误应记录服务端日志,并避免暴露堆栈、密钥或内部路径。
1 2 3 4 5 6 7 8 9 10 11
| class UserNotFoundError(ValueError): pass
@mcp.tool() def get_user(user_id: str) -> dict[str, str]: """根据用户 ID 查询用户基本信息。""" user = find_user(user_id) if user is None: raise UserNotFoundError(f"用户 {user_id} 不存在") return {"id": user.id, "name": user.name}
|
实际项目应根据 SDK 支持的工具错误机制进行映射,不要让所有异常都变成模糊的 Internal error。
不要直接暴露“万能 HTTP 工具”
旧示例中的通用 http_request(url, method, headers, ...) 存在明显风险:
- 模型可以访问内网地址或云元数据服务,造成 SSRF;
- 任意请求头可能泄漏凭据;
- 任意 HTTP 方法可能修改或删除数据;
- 返回内容可能包含提示注入或超大响应;
- 原代码还存在
retrun 拼写错误和缺少导入等问题。
更安全的方式是提供范围明确的业务工具:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24
| from urllib.parse import urljoin
import httpx from mcp.server import MCPServer
mcp = MCPServer("Order Server") BASE_URL = "https://api.example.com/"
@mcp.tool() async def get_order_status(order_id: str) -> dict[str, str]: """查询订单状态。该工具只执行只读查询。""" if not order_id.isalnum(): raise ValueError("order_id 只能包含字母和数字")
async with httpx.AsyncClient(timeout=10.0) as client: response = await client.get(urljoin(BASE_URL, f"orders/{order_id}")) response.raise_for_status() data = response.json()
return { "order_id": order_id, "status": data["status"], }
|
二、Resources:提供可寻址数据
Resource 通过 URI 暴露数据,适合文档、配置、Schema、日志摘要和知识库条目等内容。Resource 通常应是只读的,不应产生业务副作用。
固定 Resource
1 2 3 4 5 6 7 8 9
| from mcp.server import MCPServer
mcp = MCPServer("Config Server")
@mcp.resource("config://application") def get_application_config() -> str: """返回可公开给 Agent 的应用配置。""" return "environment=production\nregion=cn-north-1"
|
Resource Template
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21
| from pathlib import Path
from mcp.server import MCPServer
mcp = MCPServer("Report Server") REPORT_DIR = Path("/srv/reports").resolve()
@mcp.resource("report://{filename}") def read_report(filename: str) -> str: """读取报告目录中的 UTF-8 文本报告。""" if Path(filename).name != filename: raise ValueError("filename 不能包含目录路径")
path = (REPORT_DIR / filename).resolve() if path.parent != REPORT_DIR: raise ValueError("非法文件路径") if path.suffix not in {".txt", ".log", ".xml"}: raise ValueError("不支持的文件类型")
return path.read_text(encoding="utf-8")
|
旧版示例直接执行 open(filename),会允许调用方读取 Server 进程有权限访问的任意文件,存在路径遍历和敏感文件泄漏风险。资源 URI 是标识符,不代表它天然安全;Server 仍需执行路径白名单和访问控制。
| 需求 |
推荐 |
| 读取已知 URI 的文档或配置 |
Resource |
| 查询需要复杂参数或计算的数据 |
Tool |
| 创建、修改、删除外部数据 |
Tool |
| 列出可浏览的数据集合 |
Resource / Resource Template |
“Resource 类似 GET、Tool 类似 POST”只能作为入门类比,不能当作严格规则。只读数据库查询也可以设计成 Tool,关键在于交互语义和 Client 使用方式。
三、Prompts:可复用消息模板
Prompt 用于向用户或 Host 提供可发现、可参数化的消息模板。Prompt 通常由用户显式选择,而不是由模型自动执行。
1 2 3 4 5 6 7 8 9 10 11 12
| from mcp.server import MCPServer
mcp = MCPServer("Review Prompts")
@mcp.prompt() def review_code(language: str, focus: str = "correctness") -> str: """生成代码审查提示模板。""" return ( f"请审查下面的 {language} 代码,重点关注 {focus}。" "输出问题位置、风险、原因和修改建议。" )
|
Prompt 适合:
- 代码审查、故障分析、测试设计等固定工作入口;
- 统一团队提示词规范;
- 把常用工作流暴露给 Host 的命令面板或菜单。
Prompt 不适合存放密钥,也不应该把不可信的 Resource 内容直接拼接成高权限系统指令。
四、Context:访问请求上下文和 Client 能力
服务端处理函数可以通过 Context 访问当前请求、日志、进度通知和部分 Client 能力。不同 SDK 版本暴露的方法可能不同,以下示例使用 v2 的 MCPContext:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28
| import asyncio
from mcp.server import MCPContext, MCPServer
mcp = MCPServer("Progress Server")
@mcp.tool() async def long_running_task( task_name: str, ctx: MCPContext, steps: int = 5, ) -> str: """执行任务并报告进度。""" if not 1 <= steps <= 100: raise ValueError("steps 必须在 1 到 100 之间")
await ctx.info(f"开始执行任务:{task_name}")
for index in range(steps): await asyncio.sleep(0.1) await ctx.report_progress( progress=index + 1, total=steps, message=f"已完成 {index + 1}/{steps}", )
return f"任务 {task_name} 已完成"
|
需要注意:
- 日志、进度、Elicitation、Sampling 等能力依赖 Client 支持;
- Server 应通过能力协商或 SDK 抽象确认功能是否可用;
Context 是服务端请求上下文,不等同于 Agent 的全部对话上下文;
- 不要依赖未公开的 SDK 内部属性,它们更容易在主版本升级时变化。
五、认证与授权
MCP 的认证主要针对基于 HTTP 的远程 Server。stdio 通常依赖本机进程权限和 Host 的信任边界。
角色划分
- MCP Client:获取并携带访问令牌;
- MCP Server / Resource Server:验证令牌、受众和权限范围;
- Authorization Server:完成用户认证并签发令牌。
现代 MCP 授权建立在 OAuth 相关标准之上。生产环境不应只检查“请求头里有一个 Bearer Token”,还需要验证:
- 签名和有效期;
- 令牌受众(Audience);
- Scope / Permission;
- 令牌是否适用于当前 Resource Server;
- 当前用户是否有权执行具体工具。
Client 携带令牌
如果服务使用预先获取的静态令牌,可以在 Client 配置中传递请求头:
1 2 3 4 5 6 7 8 9 10 11 12 13
| from langchain_mcp_adapters.client import MultiServerMCPClient
client = MultiServerMCPClient( { "weather": { "transport": "http", "url": "https://mcp.example.com/mcp", "headers": { "Authorization": "Bearer YOUR_ACCESS_TOKEN", }, } } )
|
这只展示了“如何携带令牌”,不代表完整 OAuth 流程。实际项目应使用安全的凭据存储、令牌刷新和 SDK 提供的认证组件。
认证不等于授权
即使用户已经登录,也不表示其可以调用所有 Tool。Server 应在每个敏感操作上检查资源级权限。例如:
- 普通用户可以读取自己的订单,但不能读取其他用户订单;
- 测试人员可以执行测试,但不能修改生产环境配置;
- 查询工具可以自动执行,删除工具必须要求人工确认。
六、安全设计清单
- 使用最小、明确的业务工具代替通用 Shell、SQL 和 HTTP 工具;
- 校验参数长度、格式、范围和枚举值;
- 对写操作提供
dry_run、预览或人工审批;
- 设置超时、并发、速率和费用限制;
- 返回最少必要数据,避免泄漏内部字段。
Resource 安全
- 对文件目录、数据库表和 URI Scheme 使用白名单;
- 防止
../、符号链接和绝对路径绕过;
- 根据当前用户过滤 Resource 列表和内容;
- 限制响应大小,避免把超大文件直接塞入模型上下文。
Prompt 与内容安全
- 将 Resource 和 Tool 输出视为不可信输入;
- 防范提示注入,不让外部内容覆盖系统规则;
- 不在 Prompt、工具描述或日志中保存密钥;
- 对高风险调用进行二次确认。
远程服务安全
- 使用 HTTPS;
- 验证
Origin,防范 DNS rebinding;
- 本地开发服务默认绑定
127.0.0.1;
- 对认证失败、越权、工具调用和配置变更进行审计;
- Server 不得把收到的 Client Token 直接透传给下游服务。
七、完整示例
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52
| from pathlib import Path
from pydantic import BaseModel, Field from mcp.server import MCPServer
mcp = MCPServer("Testing Assistant") REPORT_DIR = Path("/srv/reports").resolve()
class Summary(BaseModel): passed: int = Field(ge=0) failed: int = Field(ge=0) pass_rate: float = Field(ge=0, le=1)
@mcp.tool() def summarize_test_result(passed: int, failed: int) -> Summary: """根据通过数和失败数计算测试通过率。""" total = passed + failed if total == 0: raise ValueError("测试用例总数不能为 0") return Summary( passed=passed, failed=failed, pass_rate=passed / total, )
@mcp.resource("report://{filename}") def read_test_report(filename: str) -> str: """读取报告目录中的测试报告。""" if Path(filename).name != filename: raise ValueError("非法文件名")
path = (REPORT_DIR / filename).resolve() if path.parent != REPORT_DIR or path.suffix != ".txt": raise ValueError("只能读取报告目录中的 .txt 文件")
return path.read_text(encoding="utf-8")
@mcp.prompt() def analyze_failure(module: str) -> str: """生成测试失败分析提示模板。""" return ( f"请分析 {module} 模块的测试失败。" "先归类错误,再给出可能原因、验证步骤和修复建议。" )
if __name__ == "__main__": mcp.run("stdio")
|
这个示例体现了三类能力的分工:
- Tool 负责计算测试指标;
- Resource 负责读取受控目录中的报告;
- Prompt 负责提供统一的失败分析入口。
小结
开发 MCP Server 时,不要只关注“能不能被模型调用”,还要关注:
- 能力应该设计成 Tool、Resource 还是 Prompt;
- 输入输出是否具有稳定 Schema;
- Server 是否正确处理错误、权限和敏感数据;
- Client 是否真的支持所使用的协议能力;
- 工具输出是否会进入模型上下文并带来提示注入风险。
MCP 提供的是标准化连接,不会自动替你完成安全设计。生产级 MCP Server 仍然需要像普通 API 一样接受认证、授权、校验、限流、监控和审计。