第5讲：MCP Server 的认证与授权

一、MCP 协议的认证缺口回顾前四讲我们实现了L1工具 description 不含内部信息L2Transport 层 TLS Origin 校验 非 root 运行L3工具参数双层校验 Resource 路径遍历防护 输出脱敏但有一个根本问题没解决谁在调用这个工具目前的 Server 对所有请求一视同仁。Agent A 查自己的订单可以Agent B 查别人的订单也可以。这在生产环境是不可接受的。MCP 协议本身不定义认证机制——它把这个责任交给了 Transport 层。对于 Streamable HTTP Transport标准做法是在 HTTP 层面做认证在应用层面做授权。二、两个身份的分离Agent Identity vs User Identity生产级 MCP 系统需要区分两个身份维度身份代表什么典型标识作用域Agent Identity​哪个 Agent/服务在发起请求Service Account / Workload ID决定“能否调用这个工具”User Identity​哪个最终用户在背后操作User Token / Session决定“操作的是谁的数据”举例CodeReviewBotAgent以sa-code-reviewproj.iam.gserviceaccount.com的身份调用query_orders_by_user工具但工具参数里的user_id必须等于当前登录用户user_zhangsan不能查别人的订单Agent Identity 管“能不能调这个工具”User Identity 管“能操作哪些数据”。三、Scope 模型设计每个工具声明自己需要的 ScopeAgent 的 Token 里携带被授予的 Scope 列表Server 端做 Scope 校验。// 工具 - Scope 映射 var toolScopes map[string]string{ query_orders_by_user: orders:read, create_refund: orders:write, send_notification: notification:send, read_resource_logs: logs:read, } // Agent Token 里的 Scope 声明 type AgentClaims struct { Sub string json:sub // Agent Identity Scopes []string json:scopes // 被授予的权限 Exp int64 json:exp }Scope 命名惯例domain:action如orders:read、orders:write、logs:read。四、Go 实现JWT Bearer Scope 校验下面实现一个带认证授权的 MCP Server支持Agent 携带 JWT Bearer Token 调用工具Server 验证 Token 签名和过期时间Server 校验 Token 的 Scope 是否覆盖工具所需的 Scope工具 handler 接收 User IdentityToken 中的sub作为参数package main import ( crypto/hmac crypto/sha256 encoding/base64 encoding/json fmt log net/http strings time ) // ---- JWT 简单实现生产环境请用 golang-jwt/jwt/v5---- type Claims struct { Sub string json:sub // Agent Identity Scopes []string json:scopes // 权限列表 Exp int64 json:exp // 过期时间 Unix 时间戳 } var jwtSecret []byte(change-me-in-production) func signJWT(claims Claims) string { header : base64.RawURLEncoding.EncodeToString([]byte({alg:HS256,typ:JWT})) payloadBytes, _ : json.Marshal(claims) payload : base64.RawURLEncoding.EncodeToString(payloadBytes) sig : hmacSHA256(jwtSecret, header.payload) return header . payload . sig } func verifyJWT(tokenStr string) (*Claims, error) { parts : strings.Split(tokenStr, .) if len(parts) ! 3 { return nil, fmt.Errorf(invalid token format) } expectedSig : hmacSHA256(jwtSecret, parts[0].parts[1]) if parts[2] ! expectedSig { return nil, fmt.Errorf(invalid signature) } payloadBytes, err : base64.RawURLEncoding.DecodeString(parts[1]) if err ! nil { return nil, fmt.Errorf(invalid payload encoding) } var claims Claims if err : json.Unmarshal(payloadBytes, claims); err ! nil { return nil, fmt.Errorf(invalid payload json) } if time.Now().Unix() claims.Exp { return nil, fmt.Errorf(token expired) } return claims, nil } func hmacSHA256(key []byte, data string) string { mac : hmac.New(sha256.New, key) mac.Write([]byte(data)) return base64.RawURLEncoding.EncodeToString(mac.Sum(nil)) } // ---- MCP 消息结构 ---- type JSONRPCReq struct { JSONRPC string json:jsonrpc ID int json:id Method string json:method Params json.RawMessage json:params,omitempty } type JSONRPCResp struct { JSONRPC string json:jsonrpc ID int json:id Result interface{} json:result,omitempty Error *RPCError json:error,omitempty } type RPCError struct { Code int json:code Message string json:message } // ---- 工具定义与 Scope 映射 ---- type ToolSpec struct { Name string json:name Description string json:description InputSchema any json:inputSchema } var toolScopes map[string]string{ query_orders_by_user: orders:read, create_refund: orders:write, send_notification: notification:send, } var toolSpecs map[string]ToolSpec{ query_orders_by_user: { Name: query_orders_by_user, Description: 按用户 ID 查询订单摘要只读, InputSchema: map[string]interface{}{ type: object, properties: map[string]interface{}{ user_id: map[string]interface{}{type: string, description: 用户 UUID}, }, required: []string{user_id}, }, }, create_refund: { Name: create_refund, Description: 发起退款申请, InputSchema: map[string]interface{}{ type: object, properties: map[string]interface{}{ order_id: map[string]interface{}{type: string}, reason: map[string]interface{}{type: string}, }, required: []string{order_id}, }, }, send_notification: { Name: send_notification, Description: 发送通知给用户, InputSchema: map[string]interface{}{ type: object, properties: map[string]interface{}{ user_id: map[string]interface{}{type: string}, message: map[string]interface{}{type: string}, }, required: []string{user_id, message}, }, }, } // ---- 带认证的 MCP Handler ---- type AuthMCPServer struct{} func (s *AuthMCPServer) Handle(claims *Claims, req JSONRPCReq) JSONRPCResp { switch req.Method { case initialize: return JSONRPCResp{ JSONRPC: 2.0, ID: req.ID, Result: map[string]interface{}{ protocolVersion: 2026-07-28, capabilities: map[string]interface{}{tools: map[string]interface{}{}}, }, } case tools/list: // 只返回 Agent 有权限的工具 var allowed []ToolSpec for name, spec : range toolSpecs { if hasScope(claims.Scopes, toolScopes[name]) { allowed append(allowed, spec) } } return JSONRPCResp{ JSONRPC: 2.0, ID: req.ID, Result: map[string]interface{}{tools: allowed}, } case tools/call: var params struct { Name string json:name Arguments json.RawMessage json:arguments } json.Unmarshal(req.Params, params) // 1. 检查 Agent 是否有该工具的 Scope requiredScope, exists : toolScopes[params.Name] if !exists { return JSONRPCResp{ JSONRPC: 2.0, ID: req.ID, Error: RPCError{Code: -32601, Message: 未知工具}, } } if !hasScope(claims.Scopes, requiredScope) { return JSONRPCResp{ JSONRPC: 2.0, ID: req.ID, Error: RPCError{Code: -32001, Message: fmt.Sprintf(权限不足需要 scope: %s, requiredScope)}, } } // 2. 执行工具简化实现 return JSONRPCResp{ JSONRPC: 2.0, ID: req.ID, Result: map[string]interface{}{ content: []map[string]interface{}{ {type: text, text: fmt.Sprintf(工具 %s 执行成功Agent: %s, params.Name, claims.Sub)}, }, }, } default: return JSONRPCResp{ JSONRPC: 2.0, ID: req.ID, Error: RPCError{Code: -32601, Message: 不支持的方法}, } } } func hasScope(scopes []string, required string) bool { for _, s : range scopes { if s required || s * { return true } } return false } // ---- HTTP Transport 带认证中间件 ---- func authMiddleware(next func(*Claims, JSONRPCReq) JSONRPCResp) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { // 从 Authorization Header 提取 Bearer Token auth : r.Header.Get(Authorization) if !strings.HasPrefix(auth, Bearer ) { http.Error(w, {jsonrpc:2.0,id:null,error:{code:-32001,message:missing authorization}}, http.StatusUnauthorized) return } tokenStr : strings.TrimPrefix(auth, Bearer ) claims, err : verifyJWT(tokenStr) if err ! nil { http.Error(w, fmt.Sprintf({jsonrpc:2.0,id:null,error:{code:-32001,message:%s}}, err.Error()), http.StatusUnauthorized) return } var req JSONRPCReq if err : json.NewDecoder(r.Body).Decode(req); err ! nil { http.Error(w, {jsonrpc:2.0,id:null,error:{code:-32700,message:parse error}}, http.StatusBadRequest) return } resp : next(claims, req) w.Header().Set(Content-Type, application/json) json.NewEncoder(w).Encode(resp) } } func main() { server : AuthMCPServer{} mux : http.NewServeMux() mux.HandleFunc(/mcp, authMiddleware(server.Handle)) // 生成测试 Token testClaims : Claims{ Sub: sa-code-reviewproj.iam.gserviceaccount.com, Scopes: []string{orders:read, notification:send}, Exp: time.Now().Add(1 * time.Hour).Unix(), } testToken : signJWT(testClaims) log.Printf(测试 Token有效期1小时:\n%s\n, testToken) log.Println(带认证的 MCP Server 启动于 :8082/mcp) log.Fatal(http.ListenAndServe(:8082, mux)) }测试# 1. 启动 Server拿到输出的测试 Token go run main.go # 2. 用有 orders:read 权限的 Token 列出工具只看到有权限的 TOKEN上面输出的 token curl -X POST localhost:8082/mcp \ -H Authorization: Bearer $TOKEN \ -H Content-Type: application/json \ -d {jsonrpc:2.0,id:1,method:tools/list} # 3. 调用有权限的工具 curl -X POST localhost:8082/mcp \ -H Authorization: Bearer $TOKEN \ -H Content-Type: application/json \ -d {jsonrpc:2.0,id:2,method:tools/call,params:{name:query_orders_by_user,arguments:{user_id:u_123}}} # 4. 调用无权限的工具create_refund 需要 orders:write应被拒绝 curl -X POST localhost:8082/mcp \ -H Authorization: Bearer $TOKEN \ -H Content-Type: application/json \ -d {jsonrpc:2.0,id:3,method:tools/call,params:{name:create_refund,arguments:{order_id:o_456}}} # 5. 不带 Token 调用应被拒绝 curl -X POST localhost:8082/mcp \ -H Content-Type: application/json \ -d {jsonrpc:2.0,id:4,method:tools/list}五、安全分层L4本讲在 L1-L3 基础上补充L4认证层Transport 层强制 Bearer Token无 Token 请求返回 401Token 必须校验签名和过期时间Token 支持 DPoP 绑定防止 Token 被盗用后在其他客户端重放生产环境用 OAuth 2.0 / OIDC 颁发 Token不用自签 HMACL4授权层每个工具声明所需 ScopeServer 在tools/call时校验tools/list只返回 Agent 有权限的工具减少模型选错的概率Scope 遵循最小权限原则不给*通配符除非管理工具写操作工具orders:write额外要求人工确认第9讲展开六、延伸阅读OAuth 2.0 for MCPdraftMCP 社区正在制定的 OAuth 2.0 授权流程草案DPoPRFC 9449Demonstration of Proof-of-Possession防止 Bearer Token 被盗用golang-jwt/jwt v5Go 生产级 JWT 库支持多种签名算法和验证回调七、下一讲预告第6讲MCP Client 架构——Agent Runtime 如何与 Server 协作前面五讲全是 Server 端视角。下一讲切换到 Client 端Agent Runtime 如何发现 Server、如何缓存工具列表、如何处理错误与重试、如何管理多个 Server 的会话生命周期。我们会用 Go 实现一个轻量 MCP Client对接我们写的 Server。开发之余的小工具推荐处理 Base64、JSON 格式化、JWT 解析、Crontab 计算、PDF 合并压缩这些碎片需求我常用一个纯前端本地工具箱zz365.top。所有计算在浏览器完成文件不上服务器关页即清。免费、无登录、无广告适合开发者当常驻标签页。