MCP协议实战：企业Agent如何安全接入内部系统

一个会出事的接法

2026 年做客服 Agent 查订单，跑通不难。把订单库连接串塞进 MCP Server 的环境变量，暴露一个 query_order 工具，参数是任意 SQL，两天就能演示。上线两周后，运维在日志里看到 Agent 执行了 SELECT * FROM users。

模型没做错，它只是在调用你给它的工具。出问题的是三处：工具边界太宽、鉴权只做到网络层、审计只记了连接数。

下面按能落地的顺序拆：架构怎么说、权限卡在哪、日志记什么、代码怎么写、云上放哪。

MCP 里谁在跟谁说话

MCP 是 Anthropic 在 2024 年 11 月开源的协议，消息体是 JSON-RPC 2.0。到本文写作时最新的规范修订版是 2025-06-18。三个角色：

Host：跑模型的宿主应用，比如 IDE、客服工作台、内部 Agent 平台。
Client：Host 内部的连接器，一条连接对应一个 Server，负责能力协商和消息路由。
Server：把内部系统能力包装成标准原语，是你要写的那部分。

Server 对外提供三类原语：tools（模型可以调用的动作）、resources（可读数据，是否放进上下文由 Host 决定）、prompts（预置提示模板）。企业内部接入绝大多数只用到 tools，先把这一个用稳。

传输层有两种，选错后面全是坑：

stdio：Server 是 Host 的子进程，身份继承父进程环境变量，没有网络层鉴权。适合单机 CLI 和本地开发工具。
Streamable HTTP：Server 是独立 HTTP 服务，替换了早期的 HTTP+SSE 方案。能挂标准中间件，能接 OAuth 2.1，能被网关和 WAF 看见。多用户、要审计、要配额的内网场景只能用这个。

一句话判断：只要调用方不止你一个人，就上 Streamable HTTP。

一次工具调用的完整链路

sequenceDiagram participant H as Host(Agent) participant C as MCP Client participant S as MCP Server participant B as 订单系统 H->>C: 用户问「订单 A123456 到哪了」 C->>S: initialize 能力协商 S-->>C: tools/list 返回工具清单 C->>S: tools/call get_order_status Note over S: 校验 token 的 aud 与 scope S->>B: 用服务账号调内网 API B-->>S: 订单数据(已按调用者过滤) S-->>C: JSON-RPC result C->>H: 结果交给模型组织语言

注意第 6 步和第 7 步。鉴权发生在 Server 收到请求时，数据过滤发生在 Server 调上游时。这两件事都在模型视野之外，模型只看到最终的 JSON。

鉴权分三层，别只做最外面那层

传输层：OAuth 2.1，token 必须绑定 audience。

MCP Server 在 OAuth 体系里是资源服务器。规范引用了 RFC 8707（Resource Indicators）和 RFC 9728（Protected Resource Metadata），核心要求是 access token 的 aud 声明白这个 token 是发给哪个 MCP Server 的。如果只写 scope 不写 aud，A 系统的 token 拿到 B 系统的地址照样能用，横向越权就是这么来的。

校验清单：签名（走 JWKS）、iss、exp、aud 等于本 Server 对外 URL。缺一项都别放行。

工具层：scope 到工具的映射，一对一。

mcp:order.read 对应 get_order_status，mcp:order.refund 对应 create_refund。别图省事发一个 mcp:* 通吃，否则只读账号能调退款。

数据层：同一个人能不能看这一条。

这层不在 MCP 规范里，必须写在工具实现内部。做法是按调用者身份做行级过滤，而不是拿服务账号的完整权限查完再返回。客服 A 只能看自己负责的店铺，这个判断得在 SQL 的 where 里，不能在返回后过滤。

还有一个容易被误用的东西：tool annotations（readOnlyHint、destructiveHint、idempotentHint）。它们只是提示，规范明确说客户端不应把 annotations 当成可信的安全依据，除非 Server 本身可信。用 readOnlyHint: true 来免掉审批流程，等于把闸门交给被调用方自己声明。

审计日志记什么、不记什么

出事后要能回答一个问题：谁在什么时间、用哪个身份、调了哪个工具、带了什么参数、拿到了什么结果、耗时多久。

记	不记
时间戳、自生成的 trace_id	token 明文、refresh token
JSON-RPC 的 request id	完整身份证号、银行卡号、手机号
调用者 `sub`、`client_id`、命中的 scope	全量返回值（除非采样）
工具名、参数摘要（脱敏/哈希）	上游系统的原始响应体
上游 HTTP 状态码、耗时、上游 trace id	数据库连接串

request id 提醒一句：JSON-RPC 的 `id` 由 Client 生成，同一次会话里可能重复，不要拿它当审计主键。自己在入口生成 UUID 贯穿到上游请求头。

高频只读工具可以做采样：1% 记录完整参数用于排障，其余记参数哈希。哈希的好处是相同参数落到同一个值，方便按频次聚类，能看出模型是不是在反复试同一个错误参数。

三类密钥分开管

混在一起是事故放大器。

Client 到 MCP Server：OAuth access token，TTL 压到 15 分钟左右。refresh token 只存在 Client 侧，Server 不落地。
MCP Server 到内部系统：服务账号凭据。放云上 KMS 或凭据管理服务，运行时拉取并缓存在内存。别写进镜像、别写进 .env 文件、别放环境变量——镜像层、kubectl describe、崩溃转储都能读到环境变量。
基础设施密钥：日志上报 AK、对象存储密钥。用实例角色（阿里云 RAM 角色、腾讯云 CAM 角色）拿 STS 临时凭据，不用长期 AK。

第 2 类里有个细节：服务账号权限按最小化配，然后在工具代码里再做一层按调用者的过滤。两层叠加，即使工具代码有 bug，服务账号本身也越不了界。

部署到阿里云还是腾讯云

两条路径都行，看你们已经在用哪套 IAM。

环节	阿里云	腾讯云
无服务器运行	函数计算 FC，支持容器镜像与自定义运行时	云函数 SCF，支持容器镜像
常驻服务	ACK、SAE	TKE、云托管
入口与鉴权	API 网关、ALB，可对接 RAM 与 JWT 鉴权插件	API 网关、CLB，可对接 CAM
密钥托管	KMS + 凭据管家	KMS + 凭据管理系统 SSM
日志与审计	日志服务 SLS	日志服务 CLS
实例身份	RAM 角色（STS 临时凭据）	CAM 角色（临时密钥）

一个实际的坑：无服务器函数有单次执行超时上限，长连接会被切断，具体数值以控制台配额页为准。MCP 的 Streamable HTTP 如果用了会话 ID 保持长连接，放 FC/SCF 上会时不时断连。两种处理方式——用无状态模式（每个请求独立、不生成 session id），或者把 Server 放到 ACK/TKE 常驻容器里。多用户生产环境我倾向后者，冷启动和连接复用都更可控。

计费维度上，FC 和 SCF 都是按调用次数加资源使用量（vCPU·秒 / GB·s）计算，单价随地域和实例规格浮动，看官网计价页最准。

Python：带鉴权的 Streamable HTTP Server

依赖：pip install 'mcp[cli]' httpx starlette uvicorn。Python SDK 迭代很快，方法名以你装到的那版 docstring 为准。

# order_mcp_server.py
import os
import uuid
import httpx
from mcp.server.fastmcp import FastMCP
from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse
from starlette.routing import Mount
UPSTREAM = os.environ['ORDER_API_BASE']  # 内网地址，由配置中心注入
mcp = FastMCP('order-gateway')
@mcp.tool()
async def get_order_status(order_id: str) -> dict:
    '''查询订单当前状态，只读。order_id 形如 A123456。'''
    if not (len(order_id) == 7 and order_id[0] == 'A' and order_id[1:].isdigit()):
        raise ValueError('order_id 格式不合法')
trace_id = str(uuid.uuid4())
    token = await fetch_service_token()  # 从 KMS/凭据管家取，内存缓存
async with httpx.AsyncClient(timeout=5) as client:
        resp = await client.get(
            f'{UPSTREAM}/orders/{order_id}',
            headers={
                'Authorization': f'Bearer {token}',
                'X-Trace-Id': trace_id,
            },
        )
    resp.raise_for_status()
    data = resp.json()
    # 只回传必要字段，别把上游响应整包塞回上下文
    return {'order_id': data['id'], 'status': data['status']}
class AuthMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        if request.url.path.startswith('/mcp'):
            header = request.headers.get('authorization', '')
            if not header.startswith('Bearer '):
                return JSONResponse({'error': 'unauthorized'}, status_code=401)
claims = verify_jwt(
                header.removeprefix('Bearer '),
                expected_aud=os.environ['MCP_PUBLIC_URL'],
            )
            if 'mcp:order.read' not in claims.get('scope', '').split():
                return JSONResponse({'error': 'forbidden'}, status_code=403)
            request.state.claims = claims
        return await call_next(request)
app = Starlette(
    routes=[Mount('/', app=mcp.streamable_http_app())],
    middleware=[Middleware(AuthMiddleware)],
)

verify_jwt 要自己实现或引库：拉 JWKS 校验签名，检查 iss、exp，并且确认 aud 等于本 Server 对外 URL。aud 这一项最容易被漏掉。

参数校验写在工具函数第一行，不要指望 Pydantic 类型注解拦住所有输入——str 类型挡不住 'A123456; DROP TABLE'。

TypeScript：无状态部署与工具注册

依赖：npm i @modelcontextprotocol/sdk express zod。

import express from 'express';
import { McpServer } from '@modelcontextprotocol/sdk/server/mcp.js';
import { StreamableHTTPServerTransport } from '@modelcontextprotocol/sdk/server/streamableHttp.js';
import { z } from 'zod';
const server = new McpServer({ name: 'order-gateway', version: '1.0.0' });
server.registerTool(
  'get_order_status',
  {
    title: '查询订单状态',
    description: '只读。入参为订单号，返回状态与更新时间。',
    inputSchema: { order_id: z.string().regex(/^A[0-9]{6}$/) },
    annotations: { readOnlyHint: true, openWorldHint: false },
  },
  async ({ order_id }) => {
    const data = await callOrderApi(order_id);
    return {
      content: [{ type: 'text', text: JSON.stringify({ order_id, status: data.status }) }],
    };
  },
);
const app = express();
app.use(express.json({ limit: '256kb' }));
app.post('/mcp', requireScope('mcp:order.read'), async (req, res) => {
  // sessionIdGenerator 传 undefined 即为无状态模式，适合函数计算部署
  const transport = new StreamableHTTPServerTransport({ sessionIdGenerator: undefined });
  res.on('close', () => { transport.close(); });
  await server.connect(transport);
  await transport.handleRequest(req, res, req.body);
});
app.listen(8080);

无状态模式下每个请求新建一次 transport，没有会话串号风险，代价是每次都要重新走一遍能力协商，延迟会略高。对内部工具类 Agent 来说这个代价通常可以接受。

开发机连内网 Server 的 Client 配置长这样：

{
  "mcpServers": {
    "order-gateway": {
      "type": "http",
      "url": "https://<你的网关域名>/mcp",
      "headers": { "Authorization": "Bearer ${ORDER_MCP_TOKEN}" }
    }
  }
}

静态 token 只适合开发机。生产环境的 Host 要走完整的 OAuth 授权码加 PKCE 流程，token 存在操作系统钥匙串或宿主的凭据存储里，不落配置文件。

十个真实会踩的坑

拿上游系统的 token 当 MCP 鉴权 token。审计上分不清是人还是服务在调，权限模型也对不上。两套 token 必须分开。
工具里直接执行原生 SQL 或 shell。参数化查询加白名单，不接受模型自由拼接。
返回值不做截断。一个订单带了 200 条物流轨迹，全塞回上下文，一次调用吃掉几万 token。分页、限长、只回必要字段。
工具数量失控。一次注册四五十个工具，模型选错概率明显上升。按场景分组，每个 Agent 只挂它需要的那几个。
靠 annotations 做安全决策。前面说过，那是提示不是契约。
Streamable HTTP 的 session id 跨用户复用。多租户下会串号。要么无状态，要么把 session 绑定到调用者身份并在不匹配时拒绝。
函数计算上强行维持长连接。超时被切断后 Client 报连接错误。换无状态模式或上常驻容器。
日志把完整参数和返回值写进 SLS/CLS。手机号、金额、地址全在里面。入库前脱敏，敏感字段只留哈希。
工具报错把堆栈原样返回。堆栈里常有内网路径、库名、版本号。对外只返回错误码加一句人话描述，细节写服务端日志。
Server 允许模型指定 URL 去拉取内容。这是 SSRF 的入口，内网元数据服务地址能被探测。目标域名走白名单，配置里写死。

上线前检查清单

每个工具的入参都有显式校验，不接受自由拼接
OAuth token 校验包含签名、iss、exp、aud 四项
scope 到工具是一对一映射，没有 mcp:*
数据层过滤写在查询条件里，不在返回后过滤
上游服务账号权限最小化，且与工具级过滤叠加
审计日志有独立 trace_id，参数已脱敏
三类密钥分存，容器环境变量里没有长期密钥
返回值有长度上限和字段白名单
错误响应不含堆栈和内部路径
部署平台的长连接超时已确认，模式选择与之匹配

明天可以先做的一件事

打开你现有的 MCP Server，导出 tools/list 的全部工具定义，把每个参数标一下：哪些能被模型自由填写，哪些会被直接拼进 SQL、文件路径、URL 或者 shell 命令。凡是命中后者的，先加白名单正则校验。这一步不动架构、不改鉴权，半小时能做完，拦住的是最常见的那类越权调用。

参考来源（均为对应厂商官方文档，内容与可用性由该站点维护）：MCP 规范 2025-06-18 修订版 https://modelcontextprotocol.io/specification/2025-06-18 ；MCP 架构说明 https://modelcontextprotocol.io/docs/concepts/architecture ；Python SDK https://github.com/modelcontextprotocol/python-sdk ；TypeScript SDK https://github.com/modelcontextprotocol/typescript-sdk ；阿里云函数计算文档 https://help.aliyun.com/zh/functioncompute/ ；腾讯云云函数文档 https://cloud.tencent.com/document/product/583 。

免责声明：文中云产品名称、规范版本号与链接均以官方站点当前内容为准，各平台配额、计费与功能可用区可能随时调整，落地前请核对控制台实际配额页。