MCP开发-08.鉴权与安全验证

发表于 | 分类于
面向未接触过鉴权的开发者，按「本地自用 → 内网共享 → 区分用户 → 统一登录 → 公网发布」的场景递进，说明 MCP 鉴权要改什么、原理穿插何处，并配示例代码。

前面我们已经介绍了怎么开发、部署一个MCP服务，并在客户端进行配置，但是会引入另外的问题。如果没办法分辨那些事合理的请求，会导致资源的滥用，甚至被不可预期的用户暴力使用，所以这个时候，我们需要开始考虑鉴权，在MCP的调用阶段进行用户的级别，确保是合法用户在使用我们的MCP服务。
目前常用的鉴权方式有 「OAuth」「Bearer」。

开始之前：鉴权到底拦的是哪一层？

想象 MCP 像一扇门后面的工具箱。Cursor 等宿主通过协议来「开门拿工具」。

  • 鉴权（authentication / authorization 在实践里常混称「鉴权」）：回答「来的人有没有资格开门、能拿哪些工具」。
  • Tool 业务参数：回答「工具怎么用」，例如 template_name=sirna-predict。

常见误区：把长期密钥做成 Tool 参数，例如 submit_template(api_token=...)。模型可能把 token 写进对话、日志，也和协议推荐的做法不一致。正确位置是：

  • 本地子进程：环境变量、本机密钥文件
  • 远程 HTTP：请求头里的凭证（后面会反复看到 Authorization: Bearer ...）

段末注释：本文说的 MCP（Model Context Protocol，模型上下文协议）指工具协议本身；Host 指 Cursor、Claude Desktop 等承载模型的应用。

场景一：只有我自己在本机 Cursor 里用

你的需求

  • 在自己电脑上调试 / 日常使用
  • 不打算给同事一个网址去连

这时候通常怎么连

Host 用 command 在本机 拉起一个 Python 进程，双方通过进程的标准输入输出说话。这种传输叫 stdio。

1
2
Cursor ──(stdin/stdout 管道)──▶ 本机 python server.py
         没有监听 0.0.0.0 端口

鉴权上你要改什么

应用层往往什么都不用加。 安全边界已经变成：

  • 谁能登录这台电脑
  • 谁能改 .cursor/mcp.json 并执行你的脚本
  • 脚本里用的本机凭据（例如 ~/.kube/config、本机 argo）属于谁。

这种模式一般不需要进行额外的鉴权，毕竟已经有权限操作服务端的电脑了，加了鉴权操作，电脑也能改

示例：最小 Server + Host 配置

1
2
3
4
5
6
7
8
9
10
11
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("local-tools")

@mcp.tool()
def ping() -> str:
    """健康检查。输入：无。输出：固定字符串。"""
    return "pong"

if __name__ == "__main__":
    mcp.run(transport="stdio")
1
2
3
4
5
6
7
8
{
  "mcpServers": {
    "local-tools": {
      "command": "/path/to/.venv/bin/python",
      "args": ["/path/to/server.py"]
    }
  }
}

场景二：想让同事用 URL 连上你的 MCP（第一次上远程）

你的需求

  • 服务跑在一台内网机器或容器里
  • 同事的 Cursor 不再执行你的 server.py，而是填写一个 url

为此我们需要做如下的步骤：

  1. Server：transport="streamable-http"，绑定合适的 host/port。
  2. 运行方式：用 systemd / Docker / 手动进程保持服务常驻；不再依赖 Cursor 拉起。
  3. 同事配置：url 指向你的 MCP 端点（常见形如 http://host:8000/mcp）。
  4. 马上要面对的问题：端口一旦对网络可达，不认识的人也能打过来。所以从下一小节开始加「门口查票」。

首先切换MCP的启动模式(stdio -> streamable-http)

本地 stdio 时，Cursor 负责启动进程（进程是在本地启动的）。
远程时，你必须先 自己把 MCP 做成常驻 HTTP 服务（类似FastAPI），Cursor 只负责连接。现行推荐传输是 Streamable HTTP（可流式的 HTTP 传输）。

1
2
3
4
【本地】Cursor  --command启动-->  python(stdio)

【远程】你先启动: python 监听 :8000/mcp
        同事 Cursor  --url-->  http(s)://那台机器/mcp

只把 mcp.run(transport="stdio") 改成 streamable-http，却仍用 command 去配 Cursor——会连不上。
启动切换后，对应的客户端配置也需要调整，进程已经在按「HTTP」听端口。远程场景下，同事的 mcp.json 应变成 url（可选再加鉴权头）。

示例：启动服务后，可以通过url访问

1
2
3
4
5
6
7
8
9
10
11
12
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("team-tools")

@mcp.tool()
def ping() -> str:
    """健康检查。"""
    return "pong"

if __name__ == "__main__":
    # 内网调试可先 127.0.0.1；要给同事机器访问再改为 0.0.0.0，并配合防火墙
    mcp.run(transport="streamable-http", host="0.0.0.0", port=8000)

同事侧：

1
2
3
4
5
6
7
{
  "mcpServers": {
    "team-tools": {
      "url": "http://10.0.0.12:8000/mcp"
    }
  }
}

我们可以先通过行面的示例，了解基于url模式的服务启动和调用模式。

场景三：内网共享，先共用一把「门口通行证」

我们之前已经开始用 http服务监听MCP的端口了，这个时候服务是在公网的，任何人都可能访问，为了安全，我们需要确保只有特定的人群可以访问我们部署到公网的服务，这个时候，我们需要开始进行鉴权。

先弄懂：Bearer 是什么

HTTP 里一种极常见的写法是：

1
Authorization: Bearer <一段字符串>

Bearer（持有者令牌）的字面意思是：谁「持有」这段字符串，谁就能用。
后面无论是「管理员发的固定口令」「JWT」还是「OAuth 换来的 access_token」，请求头长得往往一样；差别在于这段字符串 怎么发出来、服务端怎么验。

段末注释：Bearer 的用法约定见 RFC 6750。MCP 在 HTTP 资源请求里也使用该请求头形态。

鉴权上你要改什么

位置 调整
服务端 增加「读 Authorization → 和配置的密钥比对 → 失败返回 401」的逻辑（中间件或网关）
服务端配置 环境变量如 MCP_API_TOKEN（不要写进仓库）
同事 mcp.json 在 headers 里带同一把 Bearer
运维 内网 HTTPS 更佳；至少限制来源网段

依赖

  • 已使用 Streamable HTTP
  • ASGI 中间件（Starlette / FastAPI）或网关验头
  • 比对建议用 hmac.compare_digest，降低时序攻击风险

示例：单密钥校验（理解用）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
from __future__ import annotations

import hmac
import os
from typing import Callable

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import JSONResponse, Response

EXPECTED = os.environ.get("MCP_API_TOKEN", "")

class BearerAuthMiddleware(BaseHTTPMiddleware):
    """门口查票：Authorization 是否等于服务端配置的那一把密钥。"""

    async def dispatch(self, request: Request, call_next: Callable) -> Response:
        auth = request.headers.get("authorization", "")
        if not auth.lower().startswith("bearer "):
            return JSONResponse({"error": "missing bearer"}, status_code=401)
        token = auth.split(" ", 1)[1].strip()
        if not EXPECTED or not hmac.compare_digest(token, EXPECTED):
            return JSONResponse({"error": "invalid token"}, status_code=401)
        return await call_next(request)

# 将中间件挂到 FastMCP 的 Streamable HTTP ASGI app 上（具体 API 随 SDK 版本略有差异）：
# app = mcp.streamable_http_app()
# app.add_middleware(BearerAuthMiddleware)
# 再用 uvicorn 启动 app

同事配置：

1
2
3
4
5
6
7
8
9
10
{
  "mcpServers": {
    "team-tools": {
      "url": "https://mcp.internal.example/mcp",
      "headers": {
        "Authorization": "Bearer ${env:TEAM_MCP_TOKEN}"
      }
    }
  }
}

这个阶段的代价（心里有数即可）

  • 一把密钥大家共用：泄露 = 全员沦陷
  • 日志里看不出是谁调用的
  • 换密钥要通知所有人改环境变量

只适合小范围的使用，公司内部的服务功能、有限范围等不记名授权。

场景四：要分清是谁在调用，并能单独作废某人

有时候，我们可能需要知道谁在调用 MCP，并能单独废除某人的权限，记录每个人的使用情况等。这个时候，我们仍然可以在后端中间件中进行功能等扩展，引入一个用户表（记录每个人的token、和我们需要记录的其他信息）。
发钥匙时可以给 Alice、Bob 各发一串随机 token；服务端 只存哈希，明文只在发放时展示一次。吊销 = 从表删掉对应行。

1
2
3
请求: Authorization: Bearer tok_alice_xxx
服务端: sha256(tok_alice_xxx) → 查表 → user_id=alice, scopes={submit,read}
业务: 打日志、按 scopes 决定能否 submit

鉴权上你要改什么

位置 调整
服务端 密钥表（文件 / SQLite / Redis）；中间件解析 Bearer → 查表 → 注入「当前用户」
业务 Tool 需要时读取当前用户，做 scope 检查或写入审计字段
同事配置 每人 headers 里用 自己的 token，不再共用一把
管理流程 由于个人token的引入，对应增加「发钥 / 吊销」操作（哪怕先是运维手工改表）

示例：token → 用户

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
from __future__ import annotations

import hashlib
from dataclasses import dataclass
from typing import Callable

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import JSONResponse, Response

@dataclass(frozen=True)
class Principal:
    """门口验票通过后的「你是谁」。"""

    user_id: str
    scopes: frozenset[str]


_TOKEN_STORE: dict[str, Principal] = {}


def _hash_token(raw: str) -> str:
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()


def issue_token(raw_token: str, user_id: str, scopes: set[str]) -> None:
    """管理员发钥：只把 hash 与用户登记入库。"""
    _TOKEN_STORE[_hash_token(raw_token)] = Principal(user_id, frozenset(scopes))


def revoke_user(user_id: str) -> None:
    """按用户吊销：删除所有属于该用户的 hash 记录。"""
    dead = [h for h, p in _TOKEN_STORE.items() if p.user_id == user_id]
    for h in dead:
        del _TOKEN_STORE[h]


class MultiBearerMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next: Callable) -> Response:
        auth = request.headers.get("authorization", "")
        if not auth.lower().startswith("bearer "):
            return JSONResponse({"error": "missing bearer"}, status_code=401)
        raw = auth.split(" ", 1)[1].strip()
        principal = _TOKEN_STORE.get(_hash_token(raw))
        if principal is None:
            return JSONResponse({"error": "invalid token"}, status_code=401)
        request.state.principal = principal
        return await call_next(request)


def require_scope(principal: Principal, scope: str) -> None:
    """业务里二次检查权限。"""
    if scope not in principal.scopes:
        raise PermissionError(f"{principal.user_id} missing scope={scope}")

这已经能支撑「区分用户 + 吊销」。若你还希望 钥匙自动过期、票面自带角色信息，而不想维护一大张长期明文口令表，看场景五。

场景五：希望通行证会过期，并在票面上带上用户信息

希望 token 里直接带上用户 id、过期时间，MCP 本地就能验，减少长期的静态口令管理。

先弄懂：JWT 是一种「自带说明书的票」

JWT（JSON Web Token，JSON Web 令牌）通常长这样：头部.载荷.签名（三段 Base64）。
载荷里可以放 sub（主体，常是用户 id）、exp（过期时间）、aud（这张票打算给谁用）等字段。

  • 签发方用密钥给票签名
  • MCP用同一套密钥（HMAC）或公钥（RSA/EC）验签名，并检查是否过期、受众是否为自己

请求头仍然是：Authorization: Bearer eyJhbG...（看起来像乱码的 JWT）。

段末注释：载荷里的字段常叫 claim（声明）。aud（audience，受众）用来防止「给 A 服务发的票被拿到 B 服务滥用」。

鉴权上你要改什么

位置 调整
新增「签发小服务」或登录接口 用户证明身份后，签发短时 JWT（例如 1 小时）
MCP 中间件改为 jwt.decode(...)，不再做字符串全表扫描（或与表结合做黑名单）
配置 共享 JWT_SECRET（HS256）或 MCP 只持有公钥（RS256）
客户端 仍配 Bearer；只是值改成 JWT，且过期后要重新获取

注意：JWT 只解决票的格式与校验；「用户怎么登录才拿到第一张 JWT」若做成完整浏览器授权流，就会自然走到场景六的 OAuth。

示例：签发与校验（HS256）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
from __future__ import annotations

import os
import time
from typing import Any
#pip install PyJWT  # 若用 RS256，通常还需要: pip install cryptography
import jwt

JWT_SECRET = os.environ["JWT_SECRET"]
JWT_ISS = "https://auth.internal.example"
JWT_AUD = "mcp://team-tools"


def issue_access_token(user_id: str, *, ttl_sec: int = 3600) -> str:
    """登录成功后调用：签发短时访问令牌。"""
    now = int(time.time())
    payload = {
        "iss": JWT_ISS,
        "aud": JWT_AUD,
        "sub": user_id,
        "iat": now,
        "exp": now + ttl_sec,
        "scope": "submit read",
    }
    return jwt.encode(payload, JWT_SECRET, algorithm="HS256")


def verify_access_token(token: str) -> dict[str, Any]:
    """MCP 门口：验签 + 校验发行方/受众/过期。"""
    return jwt.decode(
        token,
        JWT_SECRET,
        algorithms=["HS256"],
        issuer=JWT_ISS,
        audience=JWT_AUD,
    )

场景六：同事用公司账号登录，不想私下发一堆密钥

随着业务应用场景扩大，用户增加，自己管理密钥会是一个成本极高的事情，而我们之前的应用过程中，经常看到一类模式（用微信登陆、google 登陆、github登陆）这类的方式，密钥的轮转，管理等都交给统一的账号管理体系，服务端不在进行复杂的账号管理工作。在我们的MCP鉴权中也可以使用这种方案。

先弄懂：OAuth 并不换一种请求头，它换的是「票怎么发到用户手里」，调 MCP 时，Host 带的往往 还是：

1
Authorization: Bearer <access_token>

OAuth 2.1（在 OAuth 2.0 上收紧安全实践的授权框架）规定的是：如何安全地让用户在浏览器里同意授权，并让 Host 拿到这张 access_token。

参与方可以记成三角关系：

MCP OAuth 2.1 鉴权参与方与信息流向

段末注释：IdP（Identity Provider，身份提供商）负责「你是谁」；在 OAuth 用语里它常承担 Authorization Server（授权服务器）。MCP 此时是 Resource Server（资源服务器）：不负责登录页，只验票。

授权码是管理员配好的固定值吗？

不是。 每次用户点「允许」，IdP 现场生成一次性短码（authorization code）。Host 再用这串码去换真正的 access_token。
管理员提前配置的是：应用的 Client ID（以及可选的 Client Secret）、允许的回调 URL 等——那是「登记 Cursor 这家应用」，不是给每个用户发固定业务口令。

那 IdP 要先向 MCP 领取一把「有效 MCP token」吗？

不要。 没有「第二套 MCP 预置万能密钥」需要交给 IdP。
IdP 用 自己的私钥签发 access_token；MCP 事先被配置为 信任这家 IdP（记住 issuer、JWKS 公钥地址、期望的 audience）。验签通过，就认为票有效。

1
2
管理员预置的是「信任关系」：
  MCP 知道去哪里下载 IdP 公钥，以及只接受签给自己的票

鉴权上你要改什么

位置 调整
IdP 注册应用；配置 Cursor 要求的回调地址；规划 scope
MCP 实现/挂载「按 JWKS 验 JWT」或 introspection；按规范暴露受保护资源元数据，方便客户端发现该找哪家 IdP
MCP 配置 issuer、JWKS URL、audience（绑到你的 MCP 资源标识）
同事 Cursor 使用 url；通过产品 OAuth 流程登录，或配置 auth（Client ID 等）。通常不再人手粘贴长期静态 Bearer
运维 HTTPS 必备；回调与发现端点网络可达

依赖

层级 常见选择
IdP Keycloak / Auth0 / Okta / 云厂商 IAM 等
验票 PyJWT + PyJWKClient，或网关 JWT 插件
协议细节 OAuth 2.1、PKCE、受保护资源元数据（RFC 9728）等，见文末链接

段末注释：PKCE（Proof Key for Code Exchange）用于降低授权码被截获盗用的风险，公有客户端场景下尤为重要；JWKS（JSON Web Key Set）是 IdP 发布公钥的标准方式。

示例：MCP 用 IdP 公钥验票

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
from __future__ import annotations

from typing import Any

import jwt
from jwt import PyJWKClient

ISSUER = "https://idp.example.com/realms/biocloud"
AUDIENCE = "https://mcp.example.com/sirna"
JWKS_URL = f"{ISSUER}/protocol/openid-connect/certs"

_jwks_client = PyJWKClient(JWKS_URL)


def verify_idp_token(token: str) -> dict[str, Any]:
    """验证 IdP 签发的 access_token，返回 payload（含 sub 等）。"""
    signing_key = _jwks_client.get_signing_key_from_jwt(token)
    return jwt.decode(
        token,
        signing_key.key,
        algorithms=["RS256"],
        issuer=ISSUER,
        audience=AUDIENCE,
        options={"require": ["exp", "iss", "sub"]},
    )

示例：受保护资源元数据（示意）

客户端需要知道「这个 MCP 认哪家授权服务器」。一种形态是提供 JSON 元数据（路径与发现顺序以规范为准）：

1
2
3
4
5
6
{
  "resource": "https://mcp.example.com/sirna",
  "authorization_servers": ["https://idp.example.com/realms/biocloud"],
  "scopes_supported": ["mcp:submit", "mcp:read"],
  "bearer_methods_supported": ["header"]
}

示例：Cursor 侧结构（示意）

1
2
3
4
5
6
7
8
9
10
11
12
{
  "mcpServers": {
    "argo-sirna": {
      "url": "https://mcp.example.com/sirna/mcp",
      "auth": {
        "CLIENT_ID": "${env:MCP_OIDC_CLIENT_ID}",
        "CLIENT_SECRET": "${env:MCP_OIDC_CLIENT_SECRET}",
        "scopes": ["mcp:submit", "mcp:read"]
      }
    }
  }
}

具体字段以 Cursor MCP 文档 与你所用 IdP 为准；桌面与 Web 的 OAuth 回调地址往往需要都登记到 IdP。

示例：不透明 token 时向 IdP「问询」

有的 IdP 发出的 access_token 不是 JWT，MCP 本地无法自验，可调用 introspection 接口：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
import os
import httpx

INTROSPECT_URL = "https://idp.example.com/oauth/introspect"


def introspect(token: str) -> dict:
    """问 IdP：这张票还有效吗、属于谁？"""
    r = httpx.post(
        INTROSPECT_URL,
        data={"token": token},
        auth=(os.environ["OIDC_CLIENT_ID"], os.environ["OIDC_CLIENT_SECRET"]),
        timeout=10.0,
    )
    r.raise_for_status()
    data = r.json()
    if not data.get("active"):
        raise PermissionError("token inactive")
    return data

场景七：要上公网，或公司已经有统一 API 网关

你的需求

  • 域名 + HTTPS、限流、证书，不想全塞进 Python 进程
  • 或已有 Nginx / APISIX / 云 API 网关团队规范

先弄懂：反代不是一种「新鉴权协议」

反向代理（reverse proxy）站在公网与 MCP 之间：终止 TLS、按路径转发、调超时、有时顺带验 JWT。
它解决的是「怎么安全地暴露服务」；票仍然可以是场景三～六里的任意一种。

1
2
3
Internet ──HTTPS──▶ 反代/网关 ──内网 HTTP──▶ MCP:8000
                     ↑
              证书、限流、可选验票

本机 127.0.0.1 调试：反代不是必须。
公网：强烈建议有反代（或等价的云负载均衡 + 证书）。

鉴权上你要改什么

位置 调整
DNS / 证书 为 mcp.example.com 准备 HTTPS
反代 location 转到各域 MCP 端口；流式场景注意关闭错误缓冲；长任务调大 proxy_read_timeout
鉴权策略二选一或组合 网关验票后转发；或网关只做 TLS，验票仍在 MCP 中间件
危险点 若 MCP「信任」X-User-Id 这类头，必须保证 只有网关能注入，外网不能直打 MCP 端口伪造该头

示例：Nginx 转发示意

1
2
3
4
5
6
7
8
9
10
11
12
13
server {
    listen 443 ssl;
    server_name mcp.example.com;
    # ssl_certificate ...;

    location /sirna/mcp {
        if ($http_authorization = "") { return 401; }  # 仅示意；细粒度验签常用网关插件
        proxy_pass http://127.0.0.1:8001;
        proxy_http_version 1.1;
        proxy_buffering off;
        proxy_read_timeout 3600s;
    }
}

把场景串起来：你会反复看到的同一张「票面」

从场景三到场景六，同事调 MCP 时，HTTP 层几乎都是：

1
Authorization: Bearer <这一段>
你走到的场景 <这一段> 常常是 服务端怎么认
三、共用通行证 管理员发的固定字符串 和配置相等
四、分人通行证 每人一把随机串 hash 查用户表
五、会过期的票 自签 JWT 共享密钥/公钥验签
六、公司登录 IdP 签发的 access_token（常为 JWT） JWKS / introspection

所以学鉴权时可以记一句：门上的锁孔形状往往一样（Bearer）；换场景，主要是换「谁有权印票、印完怎么验」。

对照复习：需求 → 你改哪些地方

我的情况 优先场景 Server 侧主要改动 客户端主要改动
仅本人本机 一 stdio，可不做 HTTP 鉴权 command + args
第一次给同事 URL 二 改为 Streamable HTTP 常驻 改为 url
内网、先挡住陌生人 三 Bearer 单密钥中间件 headers.Authorization
要用户身份与吊销 四 多密钥表 + 注入用户 每人不同 Bearer
要过期与自描述票 五 JWT 验签；另需签发方 Bearer 改为短时 JWT
公司账号统一登录 六 信任 IdP、验 JWKS；元数据 OAuth / auth 配置
公网或已有网关 七 置于网关后；防伪头 https:// 的 url

动手时的安全习惯（简短清单）

  • 密钥、Client Secret 不进 git
  • Token 放请求头，不放 URL 查询串
  • 生产用 HTTPS；本地调试绑定 127.0.0.1 更稳妥
  • HTTP MCP 注意 Origin 校验（防 DNS rebinding，见协议传输安全章节）
  • 静态 token 优先存哈希，并保留吊销办法
  • JWT / OAuth 务必校验过期、发行方、受众
  • 按最小权限拆 scope（只读 vs 可 submit）

延伸阅读

-------------本文结束感谢您的阅读-------------