MCP 服务器开发

MCP 工具最小实现

以下是一个完整的 MCP 工具注册示例（Python，使用 mcp SDK），包含工具注册、输入 schema 校验和三种标准返回格式：

# Python — 使用官方 mcp SDK 的最小 MCP 服务器
# pip install mcp

from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent, CallToolResult
import json

app = Server("my-mcp-server")

# 1. 注册工具列表（tools/list 响应）
@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="search_docs",
            description="按关键词搜索内部文档，返回标题与摘要。只读操作，不修改任何数据。",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "minLength": 1,
                        "maxLength": 200,
                        "description": "搜索关键词，支持中英文，勿拼接 SQL 或路径遍历字符"
                    },
                    "limit": {
                        "type": "integer",
                        "minimum": 1,
                        "maximum": 20,
                        "default": 5,
                        "description": "返回结果条数上限"
                    },
                    "category": {
                        "type": "string",
                        "enum": ["api", "guide", "faq", "all"],
                        "default": "all",
                        "description": "文档分类过滤"
                    }
                },
                "required": ["query"],
                "additionalProperties": False
            }
        )
    ]

# 2. 实现工具 handler（tools/call 响应）
@app.call_tool()
async def call_tool(name: str, arguments: dict) -> CallToolResult:
    if name == "search_docs":
        return await handle_search_docs(arguments)
    return CallToolResult(
        content=[TextContent(type="text", text=json.dumps({
            "status": "error",
            "error": {"code": "UNKNOWN_TOOL", "message": f"未知工具: {name}"}
        }))],
        isError=True
    )

async def handle_search_docs(args: dict) -> CallToolResult:
    query = args["query"]
    limit = args.get("limit", 5)
    category = args.get("category", "all")

    # 服务端二次校验（不信任模型传入的参数）
    if len(query) > 200:
        return CallToolResult(
            content=[TextContent(type="text", text=json.dumps({
                "status": "error",
                "error": {"code": "VALIDATION_ERROR", "message": "query 超过 200 字符限制"}
            }))],
            isError=True
        )

    try:
        results = await search_engine.query(query, limit=limit, category=category)
        # 成功返回格式
        return CallToolResult(
            content=[TextContent(type="text", text=json.dumps({
                "status": "success",
                "data": {"results": results, "total": len(results), "query": query}
            }))]
        )
    except TimeoutError:
        # 可重试错误
        return CallToolResult(
            content=[TextContent(type="text", text=json.dumps({
                "status": "error",
                "error": {"code": "TIMEOUT", "message": "搜索超时，可稍后重试", "retryable": True}
            }))],
            isError=True
        )

# 3. 启动 stdio 传输
async def main():
    async with stdio_server() as streams:
        await app.run(*streams)

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

TypeScript 等价实现（使用 @modelcontextprotocol/sdk）：

// TypeScript — 最小 MCP 工具实现
// npm install @modelcontextprotocol/sdk

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import {
  CallToolRequestSchema,
  ListToolsRequestSchema,
} from "@modelcontextprotocol/sdk/types.js";

const server = new Server({ name: "my-mcp-server", version: "1.0.0" }, {
  capabilities: { tools: {} },
});

server.setRequestHandler(ListToolsRequestSchema, async () => ({
  tools: [{
    name: "search_docs",
    description: "按关键词搜索内部文档，只读",
    inputSchema: {
      type: "object",
      properties: {
        query: { type: "string", minLength: 1, maxLength: 200 },
        limit: { type: "integer", minimum: 1, maximum: 20, default: 5 }
      },
      required: ["query"],
    },
  }],
}));

server.setRequestHandler(CallToolRequestSchema, async (request) => {
  if (request.params.name === "search_docs") {
    const { query, limit = 5 } = request.params.arguments as { query: string; limit?: number };
    const results = await searchDocs(query, limit);
    return {
      content: [{ type: "text", text: JSON.stringify({ status: "success", data: results }) }],
    };
  }
  throw new Error(`Unknown tool: ${request.params.name}`);
});

const transport = new StdioServerTransport();
await server.connect(transport);

传输（stdio / 远程）

stdio：子进程与宿主同机通信，适合 CLI 与桌面 Agent；注意工作目录、环境变量继承、标准输出仅用于协议行（日志应走 stderr 或专用通道），避免污染 JSON-RPC 流。

HTTP / SSE / Streamable HTTP：适合远程部署与多客户端；需 TLS、可观测的关闭与背压、以及与会话/路由匹配的 URL 设计。无论哪种传输，都应定义超时、最大消息体与并发上限，并在文档中写明宿主如何启动与回收进程。

鉴权：API key vs Bearer token 实现对比

# 方式 1：API key 在 Header（适合内部服务、简单场景）
# 客户端请求头：X-API-Key: sk-xxxx
# 服务端校验（FastAPI 示例，HTTP/SSE 传输）

from fastapi import FastAPI, Request, HTTPException
from fastapi.middleware.base import BaseHTTPMiddleware
import hmac, os

API_KEY = os.environ["MCP_API_KEY"]  # 从环境变量读取，禁止硬编码

class ApiKeyMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        key = request.headers.get("X-API-Key", "")
        # 使用 hmac.compare_digest 防止时序攻击
        if not hmac.compare_digest(key.encode(), API_KEY.encode()):
            raise HTTPException(status_code=401, detail={
                "status": "error",
                "error": {"code": "UNAUTHORIZED", "message": "无效的 API Key"}
            })
        return await call_next(request)


# 方式 2：Bearer token（OAuth 2.0，适合多租户、短期授权）
# 客户端请求头：Authorization: Bearer eyJhbGc...

import jwt  # pip install PyJWT

JWT_SECRET = os.environ["JWT_SECRET"]
JWT_ALGORITHM = "HS256"

def verify_bearer_token(authorization: str) -> dict:
    """解析并校验 JWT Bearer token，返回 payload。"""
    if not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail={
            "status": "error",
            "error": {"code": "INVALID_AUTH_SCHEME", "message": "需要 Bearer token"}
        })
    token = authorization[7:]
    try:
        payload = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM])
        return payload  # 包含 sub（用户ID）、exp（过期时间）、scope 等
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail={
            "status": "error",
            "error": {"code": "TOKEN_EXPIRED", "message": "Token 已过期，请刷新"}
        })
    except jwt.InvalidTokenError as e:
        raise HTTPException(status_code=401, detail={
            "status": "error",
            "error": {"code": "INVALID_TOKEN", "message": str(e)}
        })

# 在 tool handler 中使用 token payload 隔离租户数据
async def handle_search_docs_with_auth(args: dict, token_payload: dict) -> dict:
    tenant_id = token_payload["sub"]       # 从 token 中提取租户 ID
    scope = token_payload.get("scope", "") # 检查权限范围
    if "docs:read" not in scope:
        raise HTTPException(status_code=403, detail={
            "status": "error",
            "error": {"code": "FORBIDDEN", "message": "缺少 docs:read 权限"}
        })
    # 用 tenant_id 隔离数据查询，禁止全局查询
    return await search_engine.query(args["query"], tenant_id=tenant_id)

工具 JSON Schema 完整设计示例

完整的 JSON Schema 应包含参数类型、required、enum、description 和格式约束：

{
  "name": "create_ticket",
  "description": "创建工单。写操作，会产生数据库记录；需确认 environment 参数后再调用。",
  "inputSchema": {
    "type": "object",
    "additionalProperties": false,
    "required": ["title", "priority", "environment"],
    "properties": {
      "title": {
        "type": "string",
        "minLength": 5,
        "maxLength": 200,
        "description": "工单标题，5-200 字符，禁止 HTML 标签"
      },
      "description": {
        "type": "string",
        "maxLength": 5000,
        "description": "详细描述（可选），支持 Markdown"
      },
      "priority": {
        "type": "string",
        "enum": ["low", "medium", "high", "critical"],
        "description": "优先级。critical 会触发 on-call 通知"
      },
      "environment": {
        "type": "string",
        "enum": ["staging", "production"],
        "description": "目标环境。生产环境工单需要人工确认"
      },
      "assignee_id": {
        "type": ["string", "null"],
        "pattern": "^usr_[a-z0-9]{8}$",
        "default": null,
        "description": "负责人 ID，格式 usr_xxxxxxxx；null 表示未分配"
      },
      "due_date": {
        "type": ["string", "null"],
        "format": "date",
        "description": "截止日期，ISO 8601 格式（YYYY-MM-DD）"
      },
      "idempotency_key": {
        "type": "string",
        "minLength": 16,
        "maxLength": 64,
        "description": "幂等键，客户端生成，相同 key 不重复创建工单"
      }
    }
  }
}

三种标准返回状态（success / error / pending）的完整格式：

# success：操作完成，data 包含结果
{
  "status": "success",
  "data": {
    "ticket_id": "tkt_a1b2c3d4",
    "url": "https://app.example.com/tickets/tkt_a1b2c3d4",
    "created_at": "2026-04-11T10:30:00Z"
  }
}

# error：操作失败，Agent 根据 retryable 决定是否重试
{
  "status": "error",
  "error": {
    "code": "VALIDATION_ERROR",      # 机器可读错误码
    "message": "due_date 格式错误，需要 YYYY-MM-DD",  # 人类可读说明
    "fields": ["due_date"],           # 有问题的字段（校验类错误）
    "retryable": false                # false=不可重试（需修正参数）
  }
}

# error（可重试）：临时故障，Agent 可退避后重试
{
  "status": "error",
  "error": {
    "code": "UPSTREAM_TIMEOUT",
    "message": "下游服务超时，请在 5 秒后重试",
    "retryable": true,
    "retry_after_ms": 5000
  }
}

# pending：异步操作已接受，需轮询或等待回调
{
  "status": "pending",
  "data": {
    "job_id": "job_x9y8z7",
    "poll_url": "/api/jobs/job_x9y8z7",
    "estimated_seconds": 30
  }
}

MCP 请求主流程

下列为常见宿主与服务器之间的消息顺序示意；具体方法名以你所用 SDK 与协议版本为准。重点是：能力协商在前，列举工具在中，执行与资源访问在后。

  [ 宿主启动 MCP 服务器进程或建立远程连接 ]
        │
        ▼
  ┌─────────────────┐     交换协议版本、能力与 serverInfo；约定后续特性
  │  initialize      │     （如是否支持 tools、resources、采样等）
  └─────────────────┘
        │
        ▼
  ┌─────────────────┐     返回 prompts / resources / tools 等能力子集
  │  能力发现消息     │     （依实现可能分多条或合并）
  └─────────────────┘
        │
        ▼
  ┌─────────────────┐     模型侧需要调用工具时，宿主下发 name + 校验后参数
  │  tools/call      │──── 服务器：校验 inputSchema → 执行业务 → 结构化结果或错误
  └─────────────────┘
        │
        ▼
  ┌─────────────────┐     可选：读取资源、拉取提示模板、进度/取消通知
  │  资源 / 其它扩展  │
  └─────────────────┘
        │
        ▼
  [ 连接关闭或进程退出；清理句柄与临时凭证 ]

工具名规范化预览

工具名在清单与 tools/call 中必须一致。下方预览将输入转为小写、把非 [a-z0-9_] 转为下划线并合并连续下划线；若以数字开头则加前缀 t_。规则为教学用示例，发布前应以团队规范与宿主限制为准。