Jean's Blog

一个专注软件测试开发技术的个人博客

0%

MCP Server 开发:Tools、Resources、Prompts 与安全实践

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")

一、Tools:执行操作

Tool 用于执行计算或访问外部系统。工具可以是只读查询,也可以产生副作用,因此“工具一定有副作用”并不准确。

一个工具定义至少应该明确:

  1. 名称和描述:帮助 Host 与模型理解何时使用;
  2. 输入类型:SDK 根据类型提示生成 JSON Schema;
  3. 输出类型:让调用方可以稳定解析结果;
  4. 错误语义:区分业务失败与系统异常;
  5. 权限和副作用:明确调用是否会读取敏感数据或改变外部状态。

基础工具

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 仍需执行路径白名单和访问控制。

Resource 与 Tool 如何选择

需求 推荐
读取已知 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 应在每个敏感操作上检查资源级权限。例如:

  • 普通用户可以读取自己的订单,但不能读取其他用户订单;
  • 测试人员可以执行测试,但不能修改生产环境配置;
  • 查询工具可以自动执行,删除工具必须要求人工确认。

六、安全设计清单

Tool 安全

  • 使用最小、明确的业务工具代替通用 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 一样接受认证、授权、校验、限流、监控和审计。