Model Context Protocol
进阶特性与生产部署

进阶概览

完成 MCP 入门后，你已了解协议的基本概念与三层架构。本教程深入 MCP 的进阶特性——包括工具的错误处理与进度反馈、资源订阅、提示词模板参数化、Sampling 反向调用、传输层选择，以及在生产环境中的安全部署策略。

工具高级模式

入门阶段你已会定义 Tool 并暴露给 AI 调用。进阶模式在此基础上增加四个关键能力：错误处理、进度通知、取消支持，以及工具注解（Annotations）。

2.1 错误处理：isError vs 异常

# 正确做法：业务错误用 isError，不要抛异常
from mcp.server.fastmcp import FastMCP
from mcp.types import TextContent

mcp = FastMCP("my-server")

@mcp.tool()
async def read_file(path: str) -> str:
    try:
        with open(path) as f:
            return f.read()
    except FileNotFoundError:
        # 返回带 isError 的结果，AI 能读懂并处理
        return [TextContent(
            type="text",
            text=f"Error: file not found at {path}"
        )], True  # isError=True

// isError 让 AI 读取错误内容并决策，而非让整个调用崩溃
server.setRequestHandler(CallToolRequestSchema, async (req) => {
  if (req.params.name === "read_file") {
    try {
      const content = await fs.readFile(req.params.arguments.path, "utf-8");
      return { content: [{ type: "text", text: content }] };
    } catch (e) {
      return {
        content: [{ type: "text", text: `Error: ${e.message}` }],
        isError: true
      };
    }
  }
});

2.2 进度通知

长时间运行的 Tool（如大文件处理、批量 API 调用）可通过 progressToken 向 Client 推送进度，让用户看到实时反馈。

from mcp.server.fastmcp import Context

@mcp.tool()
async def process_large_dataset(path: str, ctx: Context) -> str:
    rows = load_csv(path)
    total = len(rows)
    for i, row in enumerate(rows):
        await process_row(row)
        # 每处理 10% 推送一次进度
        if i % (total // 10) == 0:
            await ctx.report_progress(
                progress=i,
                total=total,
                message=f"处理中 {i}/{total} 行"
            )
    return f"完成，共处理 {total} 行"

2.3 工具注解（Annotations）

注解（Annotations）是附加在工具定义上的元数据，用于向 Host 声明工具的行为特征，帮助 Host 决定是否需要用户二次确认。

资源系统

资源（Resources）是 MCP Server 向 AI 暴露的可读数据。入门阶段你了解了静态资源；进阶阶段重点是资源模板（动态 URI）和资源订阅（实时推送变化通知）。

3.1 资源模板：动态 URI

资源 URI 支持 RFC 6570 URI 模板语法，用 {variable} 表示动态参数，让一个注册规则覆盖大量同类资源。

# 注册动态资源模板：database:///{table}/{id}
@mcp.resource("database:///{table}/{id}")
async def get_record(table: str, id: str) -> str:
    # table 和 id 自动从 URI 解析注入
    record = await db.query(
        f"SELECT * FROM {table} WHERE id = ?", [id]
    )
    return json.dumps(record)

# 列出所有可用资源（支持模板展开）
@mcp.list_resources()
async def list_resources() -> list:
    tables = await db.get_tables()
    return [
        {"uri": f"database:///{t}/schema", "name": f"{t} 表结构"}
        for t in tables
    ]

3.2 资源订阅：实时推送

当资源内容发生变化时，Server 可以主动通知已订阅的 Client，Client 重新读取资源后更新上下文。适用于配置文件监听、数据库变更通知等场景。

# Server 端：当文件变化时通知所有订阅者
import asyncio
from watchfiles import awatch

async def watch_config(server):
    async for changes in awatch("/etc/myapp/config.json"):
        # 通知所有已订阅该资源的 Client
        await server.send_resource_updated(
            uri="file:///etc/myapp/config.json"
        )

提示词模板

Prompts 是 MCP Server 提供的可复用提示词模板，与 Tools 和 Resources 并列为三大原语之一。Server 注册模板，Client 按需获取并填入参数，最终得到一个可直接发送给 AI 的 messages 数组。

4.1 定义带参数的模板

@mcp.prompt()
def code_review(
    language: str,
    code: str,
    focus: str = "correctness"
) -> list:
    """代码审查模板：返回可直接发送给 AI 的 messages 数组"""
    return [
        {
            "role": "user",
            "content": {
                "type": "text",
                "text": (
                    f"请对以下 {language} 代码进行审查，"
                    f"重点关注 {focus}：

```{language}
{code}
```"
                )
            }
        }
    ]

# 带多轮对话的模板（返回 user + assistant 交替）
@mcp.prompt()
def debug_session(error_msg: str, stack_trace: str) -> list:
    return [
        {"role": "user", "content": {"type": "text", "text": f"报错信息：{error_msg}"}},
        {"role": "assistant", "content": {"type": "text", "text": "我来分析这个错误。先看堆栈："}},
        {"role": "user", "content": {"type": "text", "text": f"堆栈：{stack_trace}"}},
    ]

4.2 模板中嵌入资源

模板内容不限于纯文本——可以将 Resource 内容直接嵌入 messages，让 AI 在接收提示词的同时也获得相关数据上下文。

@mcp.prompt()
async def analyze_file(file_path: str) -> list:
    # 将文件内容作为 resource 嵌入模板
    return [
        {
            "role": "user",
            "content": [
                {"type": "text", "text": "请分析以下文件的内容和潜在问题："},
                {
                    "type": "resource",
                    "resource": {
                        "uri": f"file://{file_path}",
                        "mimeType": "text/plain"
                    }
                }
            ]
        }
    ]

Sampling 反向调用

Sampling 是 MCP 协议中最独特的能力——方向与普通 Tool 调用相反：不是 AI 调用 Server，而是 Server 向 Client 请求 AI 进行推理。这让 Server 端逻辑可以在自己的业务流程中按需嵌入 AI 能力。

5.1 Server 发起 Sampling 请求

@mcp.tool()
async def smart_summarize(url: str, ctx: Context) -> str:
    # 1. Server 先抓取页面内容
    raw_content = await fetch_page(url)

    # 2. Server 请求 AI 进行摘要（Sampling 反向调用）
    result = await ctx.sample(
        messages=[{
            "role": "user",
            "content": {
                "type": "text",
                "text": f"请用三句话概括以下内容：

{raw_content}"
            }
        }],
        max_tokens=200,
        system_prompt="你是一个简洁的摘要助手，只输出摘要，不添加其他内容。"
    )
    # 3. Server 拿到 AI 的回答，继续处理
    return result.content.text

5.2 适用场景

传输层详解

MCP 支持两种传输方式：本地进程间通信（stdio）和远程 HTTP+SSE。选择合适的传输层对安全性、延迟和部署复杂度影响显著。

6.1 stdio 模式配置（Claude Desktop）

{
  "mcpServers": {
    "my-server": {
      "command": "python",
      "args": ["-m", "my_mcp_server"],
      "env": {
        "DATABASE_URL": "postgresql://localhost/mydb",
        "API_KEY": "your-api-key-here"
      }
    }
  }
}

6.2 HTTP+SSE 模式：连接生命周期

安全最佳实践

MCP Server 有能力访问文件、数据库、API——这意味着安全设计必须从第一行代码就开始考虑，而不是上线前补丁。以下是生产级 MCP Server 的五层防御体系。

输入校验示例

from pydantic import BaseModel, field_validator
import os

class FileReadArgs(BaseModel):
    path: str
    allowed_base: str = "/home/user/workspace"

    @field_validator("path")
    def no_traversal(cls, v, values):
        base = values.data.get("allowed_base", "")
        real = os.path.realpath(v)
        # 阻止路径遍历攻击
        if not real.startswith(os.path.realpath(base)):
            raise ValueError("路径越界：不允许访问 workspace 以外的目录")
        return real

@mcp.tool()
async def read_file(path: str) -> str:
    args = FileReadArgs(path=path)  # 自动校验并抛出错误
    with open(args.path) as f:
        return f.read()

生产部署

从本地开发到生产部署，MCP Server 需要考虑容器化、配置管理、健康检查和监控。以下是标准的 Docker 部署模式。

8.1 Dockerfile（Python FastMCP）

# 使用官方 Python slim 镜像
FROM python:3.12-slim

WORKDIR /app

# 先复制依赖文件，利用 Docker 层缓存
COPY pyproject.toml .
RUN pip install --no-cache-dir fastmcp pydantic

COPY src/ ./src/

# 非 root 用户运行（最小权限原则）
RUN useradd -r -s /bin/false mcpuser
USER mcpuser

# 健康检查
HEALTHCHECK --interval=30s --timeout=10s   CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"

EXPOSE 8000
CMD ["python", "-m", "src.server"]

8.2 环境变量配置模式

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    # 所有配置从环境变量读取，带类型校验和默认值
    database_url: str
    api_key: str
    max_requests_per_minute: int = 60
    allowed_base_path: str = "/data/workspace"
    log_level: str = "INFO"

    class Config:
        env_file = ".env"  # 本地开发用 .env，生产用 Secret Manager

settings = Settings()  # 缺少必填字段时启动即失败，快速发现配置错误

8.3 关键指标监控

进阶学习

掌握 MCP 进阶特性后，以下资源可帮助你在实际项目中落地。