AI Agent 技能分享｜从零实现一个 MCP Server，让 AI Agent 安全调用内部系统

公司内部通常已经有不少接口：订单、库存、审批、客户、设备管理……把这些接口接给 Agent 并不难，难的是接完以后还能守住原来的权限边界。

如果只是把几个 HTTP 请求包成 Tool，模型确实能调用，但随之而来的问题也不少：它能看到哪些工具？用户身份从哪里来？A 租户会不会查到 B 租户的数据？写操作要不要人工确认？出了问题能不能找到是谁、在什么时间、带着什么参数调用的？

这篇从一个订单系统入手，做一个可以实际改造的 MCP Server。最终开放两个工具：

• get_order_summary：查询当前租户的订单摘要；
• create_order_hold_request：提交订单暂停申请，不直接修改订单。

第一个是只读工具，第二个有业务副作用。两者会采用不同的 Scope 和审批策略。

先定边界，再写代码

我不打算让 MCP Server 直接连订单库。内部系统既然已经有 API，权限、业务校验和审计也应该继续留在原系统中。MCP Server 更适合做一层面向 Agent 的适配：

用户
  ↓
AI Agent
  ↓  只看到允许使用的 Tool
MCP Server
  ├─ 验证访问令牌
  ├─ 检查 Scope 和租户
  ├─ 校验 Tool 参数
  ├─ 裁剪返回字段
  └─ 记录审计日志
  ↓  使用独立的服务身份
订单系统 API
  ├─ 再次检查订单归属
  ├─ 执行业务规则
  └─ 记录最终操作结果

这条链路里，模型只负责理解意图和选择工具。身份认证、数据范围和业务权限仍由代码决定。

还有一个容易混淆的地方：MCP 不是新的业务网关，也不会自动替你完成权限控制。它规定了客户端如何发现并调用 Tools、Resources 和 Prompts，但内部接口怎么鉴权、哪个用户能看哪条订单，仍然是业务系统自己的职责。

为什么不做一个万能的 request 工具

下面这种写法很省代码：

@mcp.tool()
async def request_internal_api(method: str, url: str, body: dict) -> dict:
    ...

我不会把它放进生产环境。它等于把路由选择、HTTP 方法甚至请求体结构都交给模型，原本清楚的权限边界一下变成了 Prompt 约定。

业务工具应该尽量窄。比如查询订单，就固定查询订单摘要；提交暂停申请，就固定写入申请单。工具名称、参数和返回值都应当让人一眼看明白。

OpenAI 当前的模型使用建议也提到，只暴露任务相关的工具，工具说明保持简洁、准确，并写清返回字段和错误行为。工具越多、描述越含糊，模型选错的概率通常越高。OpenAI 模型与工具使用建议

准备项目

目录不需要弄得很复杂：

internal-mcp-server/
├─ server.py
├─ agent_client.py
├─ requirements.txt
└─ .env.example

requirements.txt：

mcp[cli]>=2,<3
openai-agents
httpx>=0.28,<1
PyJWT[crypto]>=2.10,<3
pydantic>=2,<3

安装：

python -m venv .venv

# Windows
.venv\Scripts\activate

pip install -r requirements.txt

这里使用 MCP Python SDK v2。很多旧文章还在使用：

from mcp.server.fastmcp import FastMCP

v2 的服务类已经改成 MCPServer，下面的代码不要和 v1 示例混用。当前 SDK 安装与版本说明可以看 MCP Python SDK 官方文档。

.env.example：

MCP_ISSUER=https://login.example.com/
MCP_RESOURCE=http://127.0.0.1:8000/mcp
MCP_JWT_PUBLIC_KEY=-----BEGIN PUBLIC KEY-----\n请替换\n-----END PUBLIC KEY-----

INTERNAL_API_BASE_URL=https://order-api.internal
INTERNAL_API_TOKEN=请替换为服务凭据

本地可以手动设置环境变量。正式环境不要使用 .env 保存密钥，放到现有的密钥管理服务中。

访问令牌里需要什么

假设公司统一身份平台签发 JWT，MCP Server 至少会用到下面几个 Claim：

{
  "iss": "https://login.example.com/",
  "aud": "http://127.0.0.1:8000/mcp",
  "sub": "U10086",
  "client_id": "order-agent",
  "tenant_id": "T01",
  "scope": "internal:access internal:read internal:write",
  "exp": 1786600000
}

几个字段各有用途：

• iss：令牌由哪个身份平台签发；
• aud：令牌是发给哪个资源服务的；
• sub：当前用户；
• client_id：发起调用的客户端；
• tenant_id：用户所在租户；
• scope：允许使用的能力；
• exp：过期时间。

校验 JWT 时不能只验证签名。iss、aud、exp 同样要查，否则发给其他系统的 Token 也可能被拿来访问 MCP Server。MCP 的授权规范要求访问令牌绑定目标资源，并明确反对 Token passthrough，也就是不要把发给 MCP Server 的 Token 原样转交给下游 API。MCP 授权规范

本文调用订单 API 时使用单独的 INTERNAL_API_TOKEN。用户 ID 和租户 ID 只作为经过验证的业务上下文传递，订单系统还会基于这两个值重新鉴权。已有 On-Behalf-Of 或 Token Exchange 机制的公司，可以在这里换成下游 API 专用的短期 Token。

写一个 JWT TokenVerifier

新建 server.py，先把配置和 Token 校验写好：

import hashlib
import json
import logging
import os
import time
from dataclasses import dataclass
from typing import Annotated, Any

import httpx
import jwt
from jwt import PyJWTError
from mcp.server.auth.middleware.auth_context import get_access_token
from mcp.server.auth.provider import AccessToken, TokenVerifier
from mcp.server.auth.settings import AuthSettings
from mcp.server.mcpserver import Context, MCPServer
from pydantic import AnyHttpUrl, Field


logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s %(message)s",
)
logger = logging.getLogger("internal-mcp")


ISSUER = os.environ["MCP_ISSUER"]
RESOURCE = os.environ["MCP_RESOURCE"]
PUBLIC_KEY = os.environ["MCP_JWT_PUBLIC_KEY"].replace("\\n", "\n")
INTERNAL_API_BASE_URL = os.environ["INTERNAL_API_BASE_URL"]
INTERNAL_API_TOKEN = os.environ["INTERNAL_API_TOKEN"]


class JwtTokenVerifier(TokenVerifier):
    """验证由公司身份平台签发、且受众为当前 MCP Server 的 JWT。"""

    async def verify_token(self, token: str) -> AccessToken | None:
        try:
            payload = jwt.decode(
                token,
                PUBLIC_KEY,
                algorithms=["RS256"],
                issuer=ISSUER,
                audience=RESOURCE,
                options={
                    "require": ["iss", "aud", "sub", "exp"],
                },
            )
        except PyJWTError:
            logger.warning("event=token_rejected")
            return None

        raw_scope = payload.get("scope", "")
        scopes = (
            raw_scope.split()
            if isinstance(raw_scope, str)
            else list(raw_scope)
        )

        client_id = payload.get("client_id") or payload.get("azp")
        tenant_id = payload.get("tenant_id")
        subject = payload.get("sub")

        if not client_id or not tenant_id or not subject:
            return None

        return AccessToken(
            # MCP SDK 需要保存 token 字段，但业务日志绝不能输出它。
            token=token,
            client_id=str(client_id),
            scopes=scopes,
            expires_at=int(payload["exp"]),
            resource=RESOURCE,
            subject=str(subject),
            claims={
                "iss": payload["iss"],
                "tenant_id": str(tenant_id),
            },
        )

生产环境如果使用 OIDC，公钥一般来自身份平台的 JWKS。不要每次请求都临时下载 JWKS，可以按 kid 缓存公钥，并设置合理的刷新与失败策略。密钥轮换期间，新旧公钥往往需要并存一段时间。

从已验证的 Token 中拿用户身份

TokenVerifier 只负责确认“这枚 Token 是真的”。每个 Tool 还要确认“它有没有当前工具需要的 Scope”。

继续在 server.py 中加入：

@dataclass(frozen=True)
class Principal:
    user_id: str
    tenant_id: str
    client_id: str
    scopes: frozenset[str]


def require_principal(required_scope: str) -> Principal:
    token = get_access_token()
    if token is None:
        raise PermissionError("当前请求没有有效身份")

    scopes = frozenset(token.scopes)
    if required_scope not in scopes:
        raise PermissionError(f"当前用户缺少权限：{required_scope}")

    claims = token.claims or {}
    tenant_id = claims.get("tenant_id")
    if not token.subject or not tenant_id:
        raise PermissionError("访问令牌缺少用户或租户信息")

    return Principal(
        user_id=token.subject,
        tenant_id=str(tenant_id),
        client_id=token.client_id,
        scopes=scopes,
    )

tenant_id 没有出现在 Tool 参数里，这是故意的。让模型填写租户，相当于允许调用方决定自己要访问谁的数据。用户和租户必须来自已经验证的身份上下文。

封装内部订单 API

接下来写一个很薄的 HTTP 客户端。它使用 MCP Server 自己的服务凭据访问订单系统，同时把用户和租户传给下游做业务鉴权。

class InternalApiError(RuntimeError):
    pass


async def call_internal_api(
    method: str,
    path: str,
    *,
    principal: Principal,
    trace_id: str,
    json_body: dict[str, Any] | None = None,
    idempotency_key: str | None = None,
) -> dict[str, Any]:
    headers = {
        "Authorization": f"Bearer {INTERNAL_API_TOKEN}",
        "X-Actor-Id": principal.user_id,
        "X-Tenant-Id": principal.tenant_id,
        "X-Client-Id": principal.client_id,
        "X-Trace-Id": trace_id,
    }
    if idempotency_key:
        headers["Idempotency-Key"] = idempotency_key

    timeout = httpx.Timeout(
        connect=2.0,
        read=6.0,
        write=3.0,
        pool=2.0,
    )

    try:
        async with httpx.AsyncClient(
            base_url=INTERNAL_API_BASE_URL,
            timeout=timeout,
        ) as client:
            response = await client.request(
                method,
                path,
                headers=headers,
                json=json_body,
            )

        if response.status_code == 404:
            raise InternalApiError("没有找到对应记录")
        if response.status_code in {401, 403}:
            raise InternalApiError("内部系统拒绝了本次操作")

        response.raise_for_status()
        return response.json()

    except InternalApiError:
        raise
    except (httpx.TimeoutException, httpx.NetworkError):
        logger.exception(
            "event=internal_api_unavailable trace_id=%s path=%s",
            trace_id,
            path,
        )
        raise InternalApiError("内部系统暂时不可用")
    except httpx.HTTPStatusError as exc:
        logger.warning(
            "event=internal_api_error trace_id=%s path=%s status=%s",
            trace_id,
            path,
            exc.response.status_code,
        )
        raise InternalApiError("内部系统返回异常")

代码没有把下游响应体直接塞进异常。内部 API 的错误响应可能带堆栈、SQL、主机名或者调试字段，这些内容不应该回到模型上下文里。

还要注意：X-Actor-Id、X-Tenant-Id 只有在订单系统确认请求确实来自 MCP Server 时才可信。公网客户端不能直接访问订单 API，更不能自行伪造这两个请求头。

定义两个业务 Tool

现在可以创建 MCP Server 并注册工具：

mcp = MCPServer(
    "Internal Order Service",
    instructions=(
        "提供受控的订单查询和订单暂停申请能力。"
        "所有数据都受用户、租户和 Scope 限制。"
    ),
    token_verifier=JwtTokenVerifier(),
    auth=AuthSettings(
        issuer_url=AnyHttpUrl(ISSUER),
        resource_server_url=AnyHttpUrl(RESOURCE),
        # 这是进入 MCP Server 的基础 Scope。
        # read/write 仍在具体 Tool 中检查。
        required_scopes=["internal:access"],
    ),
)


def request_trace_id(ctx: Context) -> str:
    request_id = ctx.request_context.request_id
    return str(request_id) if request_id is not None else "unknown"


@mcp.tool()
async def get_order_summary(
    order_no: Annotated[
        str,
        Field(
            min_length=8,
            max_length=32,
            pattern=r"^[A-Za-z0-9_-]+$",
            description="订单编号",
        ),
    ],
    ctx: Context,
) -> dict[str, Any]:
    """查询当前用户和租户有权查看的订单摘要，不返回手机号等敏感字段。"""

    principal = require_principal("internal:read")
    trace_id = request_trace_id(ctx)
    started_at = time.perf_counter()

    try:
        data = await call_internal_api(
            "GET",
            f"/api/orders/{order_no}",
            principal=principal,
            trace_id=trace_id,
        )

        # 不把内部 API 的原始响应直接交给模型，只返回允许的字段。
        result = {
            "order_no": data["orderNo"],
            "status": data["status"],
            "customer_name": data["customerName"],
            "total_amount": data["totalAmount"],
            "created_at": data["createdAt"],
            "can_request_hold": data["canRequestHold"],
        }

        logger.info(
            "event=tool_success tool=get_order_summary trace_id=%s "
            "user_id=%s tenant_id=%s order_no=%s elapsed_ms=%.2f",
            trace_id,
            principal.user_id,
            principal.tenant_id,
            order_no,
            (time.perf_counter() - started_at) * 1000,
        )
        return result

    except InternalApiError as exc:
        logger.info(
            "event=tool_failed tool=get_order_summary trace_id=%s "
            "user_id=%s tenant_id=%s order_no=%s",
            trace_id,
            principal.user_id,
            principal.tenant_id,
            order_no,
        )
        raise RuntimeError(str(exc))


@mcp.tool()
async def create_order_hold_request(
    order_no: Annotated[
        str,
        Field(
            min_length=8,
            max_length=32,
            pattern=r"^[A-Za-z0-9_-]+$",
        ),
    ],
    reason: Annotated[
        str,
        Field(
            min_length=10,
            max_length=200,
            description="申请暂停订单的原因",
        ),
    ],
    ctx: Context,
) -> dict[str, Any]:
    """提交订单暂停申请。该工具只创建申请，不直接修改订单状态。"""

    principal = require_principal("internal:write")
    trace_id = request_trace_id(ctx)

    # 同一次业务意图得到相同的键，避免 Agent 重复提交相同申请。
    # 正式项目最好使用 Agent 运行 ID 或调用方生成的稳定业务 ID。
    raw_key = (
        f"{principal.tenant_id}:"
        f"{principal.user_id}:"
        f"hold-request:"
        f"{order_no}:"
        f"{reason.strip()}"
    )
    idempotency_key = hashlib.sha256(raw_key.encode("utf-8")).hexdigest()

    try:
        data = await call_internal_api(
            "POST",
            "/api/order-hold-requests",
            principal=principal,
            trace_id=trace_id,
            idempotency_key=idempotency_key,
            json_body={
                "orderNo": order_no,
                "reason": reason.strip(),
            },
        )

        logger.info(
            "event=tool_success tool=create_order_hold_request trace_id=%s "
            "user_id=%s tenant_id=%s order_no=%s request_id=%s",
            trace_id,
            principal.user_id,
            principal.tenant_id,
            order_no,
            data.get("requestId"),
        )

        return {
            "request_id": data["requestId"],
            "status": data["status"],
            "order_no": order_no,
        }

    except InternalApiError as exc:
        logger.info(
            "event=tool_failed tool=create_order_hold_request trace_id=%s "
            "user_id=%s tenant_id=%s order_no=%s",
            trace_id,
            principal.user_id,
            principal.tenant_id,
            order_no,
        )
        raise RuntimeError(str(exc))

这两个工具做了几件普通后端接口本来就应该做的事：

• 输入有长度和格式限制；
• 用户与租户从 Token 中取得；
• read、write 分别检查 Scope；
• 下游 API 再做订单级权限判断；
• 返回字段使用白名单；
• 写请求带幂等键；
• 错误信息经过收口；
• 日志能追到用户、租户、订单和调用耗时。

启动 Streamable HTTP 服务

在 server.py 最后加入：

if __name__ == "__main__":
    mcp.run(
        transport="streamable-http",
        host="127.0.0.1",
        port=8000,
        json_response=True,
        stateless_http=True,
    )

启动：

python server.py

默认地址是：

http://127.0.0.1:8000/mcp

示例选择 stateless_http=True，每个请求都独立处理，更适合这里的 Bearer 身份和横向扩容场景。如果你的 Tool 依赖 MCP 会话、可恢复流或服务端事件，需要根据实际需求改成有状态模式，并测试会话绑定、Token 刷新和多 Worker 行为。

远程部署时不要直接暴露开发服务器。把 ASGI App 交给现有进程管理器和网关，并配置 TLS、Host/Origin 白名单、请求体大小、连接数与超时。MCP Python SDK 默认只允许 localhost Host，换成正式域名后需要明确配置 TransportSecuritySettings。MCP Python SDK 部署说明

用 MCP Inspector 先测服务端

在接大模型之前，先用 Inspector 验证工具本身：

mcp dev server.py

连接 Streamable HTTP 地址时填写 Bearer Token，然后测试：

{
  "order_no": "O202608130001"
}

以及：

{
  "order_no": "O202608130001",
  "reason": "客户发现收货地址填写错误，申请暂停处理"
}

这里不要只看正常结果。我一般还会准备几枚不同权限的 Token：

• 没有 Token，应该返回 401；
• Token 受众不是当前 MCP Server，应该返回 401；
• 没有 internal:access，应该在 MCP 入口被拒绝；
• 有 read 没有 write，查询成功、提交申请失败；
• 租户为 T02，却查询 T01 的订单，订单系统应该返回无权访问或不存在；
• order_no 带斜杠或超长字符，应该在调用下游前被参数校验拦住。

这样可以把服务端安全和模型行为分开测试。否则 Agent 一旦回答不符合预期，很难判断究竟是模型没选对工具，还是 MCP Server 自身就有问题。

接入 AI Agent

服务端稳定以后，再写 agent_client.py：

import asyncio
import os

from agents import Agent, Runner
from agents.mcp import (
    MCPServerStreamableHttp,
    create_static_tool_filter,
)


async def main() -> None:
    async with MCPServerStreamableHttp(
        name="Internal Order MCP",
        params={
            "url": "http://127.0.0.1:8000/mcp",
            "headers": {
                "Authorization": f"Bearer {os.environ['MCP_ACCESS_TOKEN']}"
            },
            "timeout": 10,
        },
        cache_tools_list=True,
        max_retry_attempts=2,
        tool_filter=create_static_tool_filter(
            allowed_tool_names=[
                "get_order_summary",
                "create_order_hold_request",
            ]
        ),
        require_approval={
            "always": {
                "tool_names": ["create_order_hold_request"]
            },
            "never": {
                "tool_names": ["get_order_summary"]
            },
        },
    ) as mcp_server:
        agent = Agent(
            name="订单助手",
            instructions=(
                "回答订单问题前先读取订单摘要。"
                "只能依据工具返回的数据回答，不要猜测订单状态。"
                "用户要求暂停订单时，说明将提交暂停申请；"
                "不要声称订单已经暂停。"
            ),
            mcp_servers=[mcp_server],
        )

        result = await Runner.run(
            agent,
            "查一下订单 O202608130001。"
            "如果还可以处理，就申请暂停，原因是收货地址填写错误。",
        )

        if result.interruptions:
            state = result.to_state()

            for item in result.interruptions:
                print(f"等待审批的工具：{item.name}")
                print(f"调用参数：{item.arguments}")

                # Demo 中使用命令行确认。
                # 正式项目应保存 RunState，并由有权限的审批页面处理。
                answer = input("是否批准？[y/N] ").strip().lower()
                if answer in {"y", "yes"}:
                    state.approve(item)
                else:
                    state.reject(
                        item,
                        rejection_message="用户拒绝提交订单暂停申请",
                    )

            result = await Runner.run(agent, state)

        print(result.final_output)


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

启动前设置：

$env:OPENAI_API_KEY="***"
$env:MCP_ACCESS_TOKEN="***"
python agent_client.py

查询工具会直接执行，提交暂停申请则会产生 interruption，等待人工批准后才真正调用 MCP Tool。OpenAI Agents SDK 对 Streamable HTTP MCP Server 支持工具过滤、调用重试和按工具审批；官方流程是把运行结果转换为 RunState，批准或拒绝 interruption 后再恢复原 Agent。OpenAI Agents SDK：MCP、Human-in-the-loop

注意，Agent 侧的 tool_filter 和 require_approval 都不是服务端权限的替代品。攻击者可以绕过这段客户端代码直接访问 MCP Server，所以服务端仍然要检查 Token、Scope、租户和订单归属。

为什么写操作只提交申请

示例没有提供 hold_order，而是提供 create_order_hold_request。这是我在内部系统里比较喜欢的做法。

直接修改订单状态通常牵涉库存、物流、支付和审批规则。如果 MCP Tool 只是创建一张结构化申请，后续仍由原来的工作流处理，接入 Agent 时就不需要复制整套业务逻辑，风险也容易控制。

如果业务确实允许 Agent 直接执行写操作，至少要补上：

• 明确的写 Scope；
• 人工审批或额度阈值；
• 服务端再次鉴权；
• 稳定的 Idempotency-Key；
• 操作前状态检查；
• 操作后的结果核验；
• 完整的审计和补偿方案。

一句 Prompt 里的“请谨慎操作”不在这个列表里，因为它不是安全机制。

日志应该记到什么程度

排查一次 MCP 调用时，我希望能串起下面这些信息：

trace_id
client_id
user_id
tenant_id
tool_name
参数摘要
内部 API 路径
HTTP 状态码
耗时
业务结果 ID
错误类型

“参数摘要”不等于把参数完整序列化。手机号、身份证、客户地址、Token、Cookie、Authorization 请求头都不应进入日志。订单号是否需要脱敏，要按公司的数据分级规则决定。

审计日志和普通应用日志也可以分开。普通日志服务于排障，审计日志关注谁做了什么、结果如何，并且通常有更严格的访问权限和保留周期。

常见的误区

只在 Prompt 里写权限规则

Prompt 可以告诉模型什么时候使用工具，但不能证明用户有权使用工具。所有安全判断都应在模型之外重新执行。

把 MCP Token 转发给内部 API

MCP Token 的 aud 是 MCP Server，不是订单 API。原样转发既扩大了 Token 暴露范围，也破坏了受众绑定。应该换取下游专用 Token，或者使用受控的服务身份调用。

Tool 返回内部 API 的全部字段

模型只需要订单状态，不代表它也应该看到手机号、证件号和内部备注。服务端按白名单组织返回对象，别把 response.json() 原样返回。

读取和写入共用一个大 Scope

如果 internal:all 同时控制查询和写入，一旦 Token 泄露，很难限制影响范围。Scope 按能力拆分，客户端也只申请当前任务需要的部分。MCP 官方安全建议同样强调最小权限、按工具或能力拆 Scope，并禁止在日志中记录凭据。MCP 授权安全指南

自动重试所有工具

读取接口遇到 502，可以在时间预算内重试。创建申请、退款、发消息等写操作没有幂等保证时，不应自动重放。即使有幂等，也要控制总次数和总耗时，避免 Agent SDK、MCP 客户端、网关与内部 API 四层同时重试。

内部网络等于安全网络

MCP Server 部署在内网，不代表所有内网身份都应该访问它。仍然要验证 Token、限制来源、配置 TLS，并把内部 API 放在只有受信服务身份才能访问的网络边界后面。

怎么做上线前测试

第一轮不接模型，直接用 Inspector 或 MCP Client 测认证、Scope、租户隔离、参数边界和错误返回。第二轮接 Agent，重点观察自然语言有歧义时是否会选错工具、缺少参数时会不会乱填，以及拒绝审批后能否正常结束。

然后故意制造故障：订单 API 超时、返回 401、返回 500、响应缺字段、重复提交相同申请。检查 MCP Server 有没有泄露下游响应，幂等键是否稳定，日志能不能用同一个 trace_id 串起来。

最后绕过正常客户端，直接请求 MCP 地址。这一步用来确认客户端工具过滤和人工审批被绕开后，服务端权限仍然有效。只有服务端也能独立守住边界，这套接入才算可靠。

最后

把内部系统接给 Agent，不需要把原来的后端架构推倒重来。MCP Server 可以很薄：把现有 API 整理成少量业务工具，验证身份和 Scope，传递可信用户上下文，再把结果裁剪成模型需要的样子。

真正费工夫的地方仍然是熟悉的后端问题：鉴权、租户隔离、幂等、超时、审计和审批。模型只是多了一个会根据自然语言选择接口的调用方，并没有让这些问题消失。

如果只记一件事，我会选这一条：Agent 可以决定调用哪个工具，但不能决定自己拥有什么权限。