MCP 协议实战：把企业系统接进大模型工具链

工单系统、CMDB、发布平台、日志检索，这些系统都有 HTTP API。到 2026 年，让模型在对话里调这些 API 已经不算新鲜事，麻烦的是让模型自己挑工具、自己调、拿返回结果决定下一步。2024 年底之前的常规做法是给每家模型写一遍 function calling 胶水：OpenAI 一套、Claude 一套、内部自研模型再来一套，工具定义散在几个仓库里，加一个工具要改五处。

MCP 把这件事切成两半：工具方只写一个 Server，模型侧只实现一次协议。协议本体是 JSON-RPC 2.0，规范、SDK 和参考实现都在 GitHub 开源。Anthropic 在 2024 年 11 月放出第一版，之后交给社区维护，几个修订版把传输层和鉴权补齐了。

下面按接入顺序走：协议长什么样、Server 和 Client 怎么写、鉴权审计限流怎么塞进去、工具怎么设计才不被模型用坏。

示例里的数据库、IdP、域名都是占位（统一用 acme），替换成你自己的。协议细节以 `modelcontextprotocol/modelcontextprotocol` 仓库中对应版本的 schema 为准。

一、协议层要记的五个概念

MCP 的消息就是 JSON-RPC 2.0 的请求和响应，没有自定义二进制帧，没有自定义握手包。一条 tools/call 请求长这样：

{
  "jsonrpc": "2.0",
  "id": 7,
  "method": "tools/call",
  "params": {
    "name": "search_tickets",
    "arguments": { "keyword": "退款", "limit": 10 }
  }
}

服务端回：

{
  "jsonrpc": "2.0",
  "id": 7,
  "result": {
    "content": [{ "type": "text", "text": "..." }],
    "isError": false
  }
}

要记的概念只有五个：能力协商、服务端原语、客户端原语、传输、生命周期。

生命周期就四步：客户端发 initialize，服务端回 capabilities 和 serverInfo，客户端发 notifications/initialized 通知，之后正常收发。

传输只有两种要选：

二、写一个工单 Server（Python）

需要 Python 3.10 以上，示例里用到了 str | None 语法。

pip install mcp

# server.py
from contextvars import ContextVar
from datetime import datetime
from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel, Field
ASGI 中间件在鉴权通过后写入，工具函数里读它拿调用者身份
current_user: ContextVar[str] = ContextVar("current_user", default="anonymous")
mcp = FastMCP("ticket-hub", host="0.0.0.0", port=8080)
class Ticket(BaseModel):
    id: str
    title: str
    status: str = Field(description="open / pending / closed")
    owner: str
    updated_at: datetime
@mcp.tool()
def search_tickets(keyword: str, status: str | None = None, limit: int = 20) -> list[Ticket]:
    """检索当前用户有权查看的工单。
适用：用户问某个关键词相关的工单有哪些、谁在处理某类问题。
    不适用：查单张工单的完整评论和附件，那种情况用 get_ticket_detail。
Args:
        keyword: 工单标题或正文中的关键词，支持中文
        status: 可选，open / pending / closed，不传表示全部
        limit: 返回条数，服务端硬上限 50
    """
    subject = current_user.get()
    rows = ticket_db.search(
        keyword=keyword, status=status, subject=subject, limit=min(limit, 50)
    )
    return [Ticket(**r) for r in rows]
@mcp.resource("ticket://sla-rules")
def sla_rules() -> str:
    """客服 SLA 分级规则，模型回答多久算超时时读这里。"""
    return SLA_TEXT
if __name__ == "__main__":
    mcp.run(transport="streamable-http")

三件事值得说明。

函数名就是工具名，docstring 第一行是模型看到的 description，参数类型注解会被 SDK 转成 inputSchema。手写 JSON Schema 和实现不一致是常见的排查地狱，用类型注解能省掉这件事。

description 里写清「什么时候用」和「什么时候别用」，比堆参数说明有用得多。模型选工具靠的就是这段文字。只写「查询工单」，模型会在任何跟工单沾边的问题上都试一次。

current_user 这个 ContextVar 是关键。工具调下游 API 时用的是调用者身份，不是 Server 自己的服务账号。如果图省事用服务账号，模型就能读到调用者本来没权限的工单——权限过滤必须在 Server 侧做，不能靠 prompt 告诉模型「不要看别人的」。

本地调试用官方 Inspector，比直接连模型快很多：

npx @modelcontextprotocol/inspector python server.py

它能列出所有 tools、手填参数发起调用、看原始 JSON-RPC 往返消息。Server 没写对的时候，在这里排比在对话框里排省时间。

如果用 stdio 方式挂到 Claude Desktop，配置文件是这样：

{
  "mcpServers": {
    "ticket-hub": {
      "command": "python",
      "args": ["/opt/mcp/ticket_hub/server.py"],
      "env": { "TICKET_DB_DSN": "postgres://..." }
    }
  }
}

macOS 下文件在 ~/Library/Application Support/Claude/claude_desktop_config.json，Windows 下在 %APPDATA%\Claude\claude_desktop_config.json，改完要重启宿主进程。

三、Client：TypeScript 侧怎么连

npm install @modelcontextprotocol/sdk

// client.ts
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";
const transport = new StreamableHTTPClientTransport(
  new URL("https://mcp.internal.acme/ticket-hub"),
  // 显式写死 token 只适合内部调试，生产走 OAuth 授权流
  { requestInit: { headers: { Authorization: Bearer ${process.env.MCP_TOKEN} } } },
);
const client = new Client({ name: "ticket-agent", version: "0.1.0" });
await client.connect(transport);
const { tools } = await client.listTools();
console.log(tools.map((t) => t.name));
const result = await client.callTool({
  name: "search_tickets",
  arguments: { keyword: "退款", limit: 10 },
});
// result.content 是内容数组，result.isError 表示工具层的失败
console.log(result.content);
await client.close();

这里有个容易混的点：协议层错误和工具层错误不是一回事。参数不符合 schema、工具名不存在，走 JSON-RPC 的 error 字段；工具执行了但业务上失败（比如工单不存在），走 result.isError: true，错误文字会喂回模型。多数宿主更愿意处理第二种，模型看到文字能自己换策略。

四、鉴权：OAuth 2.1 加 Resource Indicators

远程 Server 用静态 token 分发给全公司，等于没有鉴权。MCP 的授权规范把 Server 定义成 OAuth 2.1 的 Resource Server，客户端从企业 IdP 拿 token。

流程是这样：

• 客户端第一次请求没带 token，Server 返回 401，带上 WWW-Authenticate: Bearer resource_metadata="https://mcp.internal.acme/.well-known/oauth-protected-resource"。
• 客户端读 Protected Resource Metadata（RFC 9728），知道授权服务器是谁。
• 客户端走授权码加 PKCE，token 请求里带 resource=https://mcp.internal.acme/ticket-hub，这是 RFC 8707 的 Resource Indicators。
• 拿到 access token，之后每个请求带 Authorization: Bearer。

import jwt
from jwt import PyJWKClient
jwks = PyJWKClient("https://idp.internal.acme/.well-known/jwks.json")
MCP_RESOURCE = "https://mcp.internal.acme/ticket-hub"
def verify(token: str) -> dict:
    key = jwks.get_signing_key_from_jwt(token).key
    return jwt.decode(
        token,
        key,
        algorithms=["RS256"],
        audience=MCP_RESOURCE,   # 必须和 token 请求里 resource 参数一致
        issuer="https://idp.internal.acme/",
        options={"require": ["exp", "iat", "aud", "iss"]},
    )

一条硬规则：不要把客户端带来的 token 原样转发给下游 API。MCP 规范明确禁止 token passthrough。正确做法是用 Server 自己的服务身份调下游，把调用者身份作为参数传下去做权限裁剪；或者做 token exchange，换一个 audience 指向下游的 token。转发原始 token 的问题是审计上根本分不清是哪个服务在用这个凭证。

五、审计：每次工具调用落一条能查的记录

审计要能回答四个问题：谁、什么时候、调了什么、结果如何。含混的日志在事故复盘时帮不上忙。

import hashlib
import json
import logging
import sys
import time
audit = logging.getLogger("mcp.audit")
audit.setLevel(logging.INFO)
stdio 传输下 stdout 是 JSON-RPC 通道，日志只能往 stderr 或文件走
audit.addHandler(logging.StreamHandler(sys.stderr))

def audit_call(subject, session_id, tool, arguments, started, result, error=None):
    params = json.dumps(arguments, sort_keys=True, ensure_ascii=False)
    audit.info(json.dumps({
        "ts": round(time.time(), 3),
        "subject": subject,
        "session_id": session_id,
        "tool": tool,
        "params_digest": hashlib.sha256(params.encode()).hexdigest(),
        "params_keys": sorted(arguments.keys()),
        "duration_ms": round((time.time() - started) * 1000, 1),
        "status": "error" if error else "ok",
        "error_code": error,
        "result_bytes": len(json.dumps(result, ensure_ascii=False)) if result else 0,
    }, ensure_ascii=False))

params 只落摘要和字段名，是有意的。工单正文、用户手机号这类内容进了日志，日志平台就成了新的泄露面。需要还原参数时用 subject + ts 去业务系统侧对，别指望审计日志本身。

六、限流：按人、按工具两个维度

为什么是两个维度：一个用户狂点搜索是人的问题；一个工具被所有人高频调用，通常是工具设计的锅，多半是 description 写得太宽，模型什么都往里塞。

import threading
import time
class TokenBucket:
    def __init__(self, rate: float, burst: int):
        self.rate, self.burst = rate, burst
        self._state: dict[str, tuple[float, float]] = {}
        self._lock = threading.Lock()
def allow(self, key: str) -> bool:
        now = time.monotonic()
        with self._lock:
            tokens, last = self._state.get(key, (float(self.burst), now))
            tokens = min(self.burst, tokens + (now - last) * self.rate)
            if tokens < 1:
                self._state[key] = (tokens, now)
                return False
            self._state[key] = (tokens - 1, now)
            return True
按人和按工具各一个桶，key 形如 "u:alice:search_tickets"
limiter = TokenBucket(rate=0.5, burst=10)  # 每秒 0.5 次，突发 10 次

几个实操点：

内存版够用于 stdio 传输和单实例部署。多实例要换 Redis，用 Lua 脚本或 INCR 加 EXPIRE 保证原子性，两个进程各扣一次会超发。

触发限流的表达方式看场景。恶意刷量在 Streamable HTTP 层直接回 429 加 Retry-After；如果是正常的用量超限，把说明作为 isError: true 的工具结果返回，模型看到文字会自己换策略或告诉用户，比直接断流体验好——直接断流模型会反复重试。

要在协议层表达时，自定义错误码放在 -32000 到 -32099 区间，这段是 JSON-RPC 留给实现方定义的。

七、工具编排：6 个任务级工具好过 40 个接口级工具

把一个 REST API 的每个端点包成一个工具，是最容易走的路，也是最快翻车的路。工具列表超过 20 个之后，模型选错的概率明显上升，工具定义本身也把上下文吃掉一大块。

• 参数控制在 5 个以内。超过 5 个通常说明这个工具该拆。
• 默认分页，默认返回 10 到 20 条，给模型 limit 参数但服务端硬截断（比如 50），别让客户端决定上限。
• 写操作单独成工具，名字里带动词，description 里写明会修改数据、调用前必须向用户确认。
• 不在工具里做多步业务编排。Server 只做一件事，编排交给模型或宿主侧的 agent loop，否则工具会变成一个个隐藏的业务流程，出问题只能靠猜。

代价是工具名冲突要加前缀，各 Server 的 capabilities 要合并，Streamable HTTP 的会话状态要处理。团队规模小的时候，客户端直连几个 Server 反而更省事，不必为了架构好看先上网关。

八、企业落地：客服团队接工单系统的推进顺序

场景很具体：三线支持在对话框里问「上周华东区退款类工单里，有哪些卡在等待财务审批超过 3 天」。

• 只读工具。先做 search_tickets 和 get_ticket_detail，全部只读，权限过滤走调用者身份。
• 身份打通。OAuth 接企业 IdP，校验 aud 和 scope，审计日志落地并且能被检索。
• 限流与配额。按 subject 和工具做配额，先宽后紧，观察两周用量分布再收紧。
• 写操作。update_ticket_status、add_ticket_comment 上线，同步加「调用前需用户确认」的描述，加审计告警，比如单用户 5 分钟内修改超过 20 张工单就报警。
• 资源与提示词。把工单模板、话术、SLA 规则做成 resources 和 prompts，省掉模型每次都重新拼的上下文。

九、踩坑清单

• stdio Server 用 print 打日志。stdout 是 JSON-RPC 通道，会被污染，症状是初始化直接失败或工具偶发无响应。日志走 stderr 或文件。
• 工具 description 只写「查询工单」。模型不知道什么时候该用它。写清触发场景和不该用的场景。
• 手写 JSON Schema。用类型注解或 zod，schema 和实现不一致时排查成本极高。
• 一次返回 500 条工单。上下文直接爆掉，模型开始丢字段。服务端分页加硬截断。
• 用服务账号调下游。所有人看到一样的数据，权限体系形同虚设。
• 校验了签名没校验 aud。给别的服务签的 token 也能用，也就是 confused deputy。
• 把客户端 token 转发给下游 API。规范禁止，审计上也没法区分是谁在用。
• 新项目还用 HTTP+SSE 传输。2025-06-18 版规范已标为弃用，直接用 Streamable HTTP。
• 忘了处理 Mcp-Session-Id。Streamable HTTP 是有会话的，丢了 header 会表现为偶发的工具调用失败，很难复现。
• 反向代理开了响应缓冲。长时间运行的工具调用被 Nginx 攒住，客户端先超时。
• 工具报错把 stack trace 返回给模型。里面可能带内部表名、SQL、密钥路径。错误信息要区分「给用户看的」和「给日志看的」。
• 敏感参数原文写进审计日志。落摘要和字段名足够排查了。

十、参考仓库

• MCP 规范与 schema：https://github.com/modelcontextprotocol/modelcontextprotocol
• Python SDK（FastMCP）：https://github.com/modelcontextprotocol/python-sdk
• TypeScript SDK：https://github.com/modelcontextprotocol/typescript-sdk
• 官方参考 Server 集（文件系统、Git、数据库等）：https://github.com/modelcontextprotocol/servers
• 调试工具 MCP Inspector：https://github.com/modelcontextprotocol/inspector

下一步

挑一个只读接口，用 Python SDK 包成 3 个任务级工具，跑 npx @modelcontextprotocol/inspector python server.py 手动调一遍。能被 Inspector 正确列出工具、参数校验符合预期、错误分支返回的是给模型看的文字而不是 stack trace——这三条过了，再谈接模型。

免责声明：文中的 OAuth 校验、限流、审计实现均为最小示例，生产使用前需按企业安全规范评审。提到的第三方仓库与工具版权归各自所有者，链接内容可能随版本迭代变化，以仓库当前状态为准。