给 MCP 服务器加上鉴权

这篇适合谁，读完能做到什么

MCP 服务器只要能被访问，就要决定谁能连、连上后能做什么。这篇分三步讲：先用 Authorization 请求头传一个简单凭证，再换成签名的 JWT，最后用中间件做按角色的权限检查。读完你能说清每一步挡住了什么、还缺什么，以及下一步该看什么^[1]。

鉴权（authentication）和授权（authorization）要分开看

鉴权回答的是：这个人能不能进到放着 MCP 服务器功能的资源服务器。授权回答的是：这个人能不能用某一个具体资源，比如能读内容但不能删除^[1]。

这门课还提到，MCP SDK 支持 OAuth 2.1，它涉及授权服务器、资源服务器和 bearer token 的换取流程，比较复杂。新手建议先从基础的鉴权做起，再逐步加强^[1]。

第一步：用 Authorization 请求头传凭证

基础凭证通常是用 base64 编码的用户名和密码，或者一个 API key，放在 Authorization 请求头里发送^[1]。中间件是在每个请求中间运行的代码：它检查凭证，凭证有效就放行，否则给客户端返回鉴权错误^[1]。

Python 示例里，请求头缺失时返回 401 Unauthorized，token 无效时返回 403 Forbidden^[1]：

class AuthMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):

        has_header = request.headers.get("Authorization")
        if not has_header:
            print("-> Missing Authorization header!")
            return Response(status_code=401, content="Unauthorized")

        if not valid_token(has_header):
            print("-> Invalid token!")
            return Response(status_code=403, content="Forbidden")

        print("Valid token, proceeding...")
        print(f"-> Received {request.method} {request.url}")
        response = await call_next(request)
        response.headers['Custom'] = 'Example'
        return response

示例里的 valid_token 函数注明了不能用于生产环境：

# DON'T use for production - improve it !!
def valid_token(token: str) -> bool:
    # remove the "Bearer " prefix
    if token.startswith("Bearer "):
        token = token[7:]
        return token == "secret-token"
    return False

这门课明确说，密钥不应该写在代码里。用来比对的值应该来自数据源或身份提供方，或者干脆让身份提供方来做校验^[1]。

TypeScript 用 Express 做同样的事。请求头缺失时返回 401，isValid 检查不通过时返回 403，否则调用 next() 放行^[1]：

app.use((req, res, next) => {
    console.log('Request received:', req.method, req.url, req.headers);
    console.log('Headers:', req.headers["authorization"]);
    if(!req.headers["authorization"]) {
        res.status(401).send('Unauthorized');
        return;
    }
    
    let token = req.headers["authorization"];

    if(!isValid(token)) {
        res.status(403).send('Forbidden');
        return;
    }  

    console.log('Middleware executed');
    next();
});

客户端这边，Python 示例把凭证放进请求头发出去，并提醒不要把值硬编码在代码里^[1]：

# DON'T hardcode the value, have it at minimum in an environment variable or a more secure storage
token = "secret-token"

async with streamablehttp_client(
        url = f"http://localhost:{port}/mcp",
        headers = {"Authorization": f"Bearer {token}"}

TypeScript 客户端把请求头放在 requestInit 里，再把这个 options 对象传给 transport^[1]：

// DON'T hardcode the value like shown here. At minimum have it as a env variable and use something like dotenv (in dev mode).
let token = "secret123"

// define a client transport option object
let options: StreamableHTTPClientTransportOptions = {
  sessionId: sessionId,
  requestInit: {
    headers: {
      "Authorization": "secret123"
    }
  }
};

这种做法适合很简单、不允许未鉴权调用的接口，算是个好的起点。但只要稍复杂一点就有风险：凭证可能被偷走，即使用 HTTPS 也一样。所以还需要能撤销 token 的机制，以及按来源地、请求频率等的额外检查^[1]。

第二步：换成 JWT

JWT 由三部分组成：说明算法和类型的头部（header）、sub、exp、role 等声明（claims）组成的载荷（payload），以及用密钥或私钥签的签名（signature）^[1]。它的好处是 token 有过期时间，角色、scope 和权限可以做细粒度控制，token 自带信息所以服务器不用保存会话，还能在本地校验^[1]。

Python 示例先构造 header 和 payload，再用 jwt.encode 编码^[1]：

secret_key = 'your-secret-key'

header = {
    "alg": "HS256",
    "typ": "JWT"
}

# the user info and its claims and expiry time
payload = {
    "sub": "1234567890",               # Subject (user ID)
    "name": "User Userson",                # Custom claim
    "admin": True,                     # Custom claim
    "iat": datetime.datetime.utcnow(),# Issued at
    "exp": datetime.datetime.utcnow() + datetime.timedelta(hours=1)  # Expiry
}

# encode it
encoded_jwt = jwt.encode(payload, secret_key, algorithm="HS256", headers=header)

TypeScript 示例先安装依赖^[1]：

npm install jsonwebtoken
npm install --save-dev @types/jsonwebtoken

然后用 jwt.sign 生成 token。示例里的 token 用 HS256 签名，有效期一小时^[1]：

const token = jwt.sign(payload, secretKey, {
  algorithm: 'HS256',
  header: header
});

服务器收到 token 后要校验。Python 用 jwt.decode(token, secret_key, algorithms=["HS256"])，示例里用 ExpiredSignatureError 和 InvalidTokenError 捕获失败情况^[1]。TypeScript 用 jwt.verify，token 结构不对或已失效时它会报错^[1]：

try {
  const decoded = jwt.verify(token, secretKey);
  console.log('Decoded Payload:', decoded);
} catch (err) {
  console.error('Token verification failed:', err);
}

校验还应该确认，token 指向的是你系统里真实存在的用户，并且这个用户确实有 token 声明的权限^[1]。

第三步：按角色控制权限（RBAC）

示例的 RBAC 有三个权限等级：Admin.Write、User.Read 和 Guest.Read。管理员什么都能做，普通用户能读写，访客只能读^[1]。下面的 Python 中间件检查解码后 token 里的 permissions 列表，是否包含 REQUIRED_PERMISSION（示例中设为 User.Read）。不包含就返回 403 和 Permission denied^[1]：

class JWTPermissionMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        auth_header = request.headers.get("Authorization")
        if not auth_header or not auth_header.startswith("Bearer "):
            return JSONResponse({"error": "Missing or invalid Authorization header"}, status_code=401)

        token = auth_header.split(" ")[1]
        try:
            decoded = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        except jwt.ExpiredSignatureError:
            return JSONResponse({"error": "Token expired"}, status_code=401)
        except jwt.InvalidTokenError:
            return JSONResponse({"error": "Invalid token"}, status_code=401)

        permissions = decoded.get("permissions", [])
        if REQUIRED_PERMISSION not in permissions:
            return JSONResponse({"error": "Permission denied"}, status_code=403)

        request.state.user = decoded
        return await call_next(request)

怎么给 MCP 的每个功能加权限

这门课说，中间件很难按 MCP 的单个功能做权限控制。它建议的办法是：在每个工具、资源或提示词里单独检查权限；或者用高级服务器写法里的请求处理器，把检查集中到更少的地方^[1]。示例的 Python has_permission 只要用户拥有所需权限中的任意一个就返回真。它把 create_product 对应到 User.Write 或 Admin.Write，把 delete_product 对应到 Admin.Write^[1]。

限制和常见问题

示例里的 secret-token、your-secret-key 都是占位值，真实项目要放在安全的地方，例如环境变量^[1]。
Express 有现成的辅助库：express-jwt 负责解码 token 并挂到 req.user 上，express-jwt-permissions 提供 guard 中间件检查权限^[1]。