MCP 2025-03-26 新规范解读：OAuth认证、远程连接与Tool Annotations

POST /register HTTP/1.1
Content-Type: application/json

{
  "client_name": "My AI Assistant",
  "redirect_uris": ["http://localhost:3000/callback"],
  "grant_types": ["authorization_code"],
  "token_endpoint_auth_method": "none"
}

import express from 'express';
import crypto from 'crypto';
import jwt from 'jsonwebtoken';

const app = express();
app.use(express.json());

// --- OAuth 2.1 端点 ---

// 授权服务器元数据发现
app.get('/.well-known/oauth-authorization-server', (req, res) => {
  res.json({
    issuer: 'https://mcp.example.com',
    authorization_endpoint: 'https://mcp.example.com/authorize',
    token_endpoint: 'https://mcp.example.com/token',
    registration_endpoint: 'https://mcp.example.com/register',
    response_types_supported: ['code'],
    code_challenge_methods_supported: ['S256'],
    grant_types_supported: ['authorization_code', 'refresh_token']
  });
});

// 动态客户端注册 (RFC 7591)
const clients = new Map();
app.post('/register', (req, res) => {
  const clientId = crypto.randomUUID();
  clients.set(clientId, {
    client_name: req.body.client_name,
    redirect_uris: req.body.redirect_uris,
    created_at: Date.now()
  });
  res.status(201).json({ client_id: clientId });
});

// Token 端点 (验证 PKCE)
app.post('/token', (req, res) => {
  const { code, code_verifier, client_id } = req.body;
  const stored = authCodes.get(code);
  if (!stored) return res.status(400).json({ error: 'invalid_grant' });

  // 验证 PKCE: SHA256(code_verifier) === code_challenge
  const challenge = crypto
    .createHash('sha256')
    .update(code_verifier)
    .digest('base64url');
  if (challenge !== stored.code_challenge) {
    return res.status(400).json({ error: 'invalid_grant' });
  }

  const accessToken = jwt.sign(
    { sub: stored.user_id, client_id, scope: stored.scope },
    process.env.JWT_SECRET,
    { expiresIn: '15m' }
  );
  const refreshToken = crypto.randomUUID();

  res.json({
    access_token: accessToken,
    token_type: 'Bearer',
    expires_in: 900,
    refresh_token: refreshToken
  });
});

// --- MCP Streamable HTTP 端点 ---
app.post('/mcp', authenticateToken, (req, res) => {
  // 处理 JSON-RPC 请求（含批处理）
  const body = req.body;
  if (Array.isArray(body)) {
    // 批处理模式
    const results = body.map(msg => handleJsonRpc(msg, req.user));
    res.json(results.filter(r => r !== null));
  } else {
    const result = handleJsonRpc(body, req.user);
    result ? res.json(result) : res.status(202).end();
  }
});

function authenticateToken(req, res, next) {
  const token = req.headers.authorization?.split(' ')[1];
  if (!token) return res.status(401).json({ error: 'unauthorized' });
  try {
    req.user = jwt.verify(token, process.env.JWT_SECRET);
    next();
  } catch {
    res.status(403).json({ error: 'invalid_token' });
  }
}

import hashlib
import base64
import os
import httpx

class MCPOAuthClient:
    def __init__(self, server_url: str):
        self.server_url = server_url
        self.client_id = None
        self.access_token = None

    async def discover(self):
        """发现授权服务器元数据"""
        async with httpx.AsyncClient() as client:
            resp = await client.get(
                f"{self.server_url}/.well-known/oauth-authorization-server"
            )
            return resp.json()

    async def register(self):
        """动态客户端注册"""
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                f"{self.server_url}/register",
                json={
                    "client_name": "Python MCP Client",
                    "redirect_uris": ["http://localhost:8080/callback"],
                    "grant_types": ["authorization_code"]
                }
            )
            self.client_id = resp.json()["client_id"]

    def generate_pkce(self) -> tuple[str, str]:
        """生成 PKCE code_verifier 和 code_challenge"""
        code_verifier = base64.urlsafe_b64encode(
            os.urandom(32)
        ).decode('utf-8').rstrip('=')
        code_challenge = base64.urlsafe_b64encode(
            hashlib.sha256(code_verifier.encode()).digest()
        ).decode('utf-8').rstrip('=')
        return code_verifier, code_challenge

    async def exchange_token(self, code: str, code_verifier: str):
        """使用授权码 + code_verifier 换取 Access Token"""
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                f"{self.server_url}/token",
                data={
                    "grant_type": "authorization_code",
                    "code": code,
                    "code_verifier": code_verifier,
                    "client_id": self.client_id
                }
            )
            tokens = resp.json()
            self.access_token = tokens["access_token"]
            return tokens

    async def call_tool(self, tool_name: str, arguments: dict):
        """调用 MCP 工具"""
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                f"{self.server_url}/mcp",
                headers={"Authorization": f"Bearer {self.access_token}"},
                json={
                    "jsonrpc": "2.0",
                    "id": 1,
                    "method": "tools/call",
                    "params": {
                        "name": tool_name,
                        "arguments": arguments
                    }
                }
            )
            return resp.json()

POST /mcp HTTP/1.1
Content-Type: application/json
Accept: application/json, text/event-stream
Mcp-Session-Id: sess_abc123

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": { "name": "long_running_task", "arguments": {} }
}

HTTP/1.1 200 OK
Content-Type: application/json

{"jsonrpc": "2.0", "id": 1, "result": {"content": [{"type": "text", "text": "done"}]}}

HTTP/1.1 200 OK
Content-Type: text/event-stream

event: message
data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"progress":50,"message":"处理中..."}}

event: message
data: {"jsonrpc":"2.0","id":1,"result":{"content":[{"type":"text","text":"完成"}]}}

// 客户端断线重连逻辑
async function reconnect(sessionId, lastEventId) {
  const response = await fetch('/mcp', {
    method: 'GET',
    headers: {
      'Accept': 'text/event-stream',
      'Mcp-Session-Id': sessionId,
      'Last-Event-ID': lastEventId
    }
  });
  // Server 从 lastEventId 之后继续推送事件
  const reader = response.body.getReader();
  // ... 处理 SSE 事件流
}

interface ToolAnnotations {
  title?: string;              // 语义化标题（UI展示用）
  readOnlyHint?: boolean;      // 是否只读（默认 false）
  destructiveHint?: boolean;   // 是否有破坏性（默认 true）
  idempotentHint?: boolean;    // 是否幂等（默认 false）
  openWorldHint?: boolean;     // 是否与外部交互（默认 true）
}

{
  "tools": [
    {
      "name": "query_database",
      "description": "Execute a read-only SQL query",
      "inputSchema": {
        "type": "object",
        "properties": {
          "sql": { "type": "string" }
        }
      },
      "annotations": {
        "title": "Database Query",
        "readOnlyHint": true,
        "destructiveHint": false,
        "idempotentHint": true,
        "openWorldHint": false
      }
    },
    {
      "name": "delete_user",
      "description": "Permanently delete a user account",
      "inputSchema": {
        "type": "object",
        "properties": {
          "user_id": { "type": "string" }
        }
      },
      "annotations": {
        "title": "Delete User Account",
        "readOnlyHint": false,
        "destructiveHint": true,
        "idempotentHint": false,
        "openWorldHint": false
      }
    },
    {
      "name": "send_email",
      "description": "Send an email to the specified recipient",
      "inputSchema": {
        "type": "object",
        "properties": {
          "to": { "type": "string" },
          "subject": { "type": "string" },
          "body": { "type": "string" }
        }
      },
      "annotations": {
        "title": "Send Email",
        "readOnlyHint": false,
        "destructiveHint": false,
        "idempotentHint": false,
        "openWorldHint": true
      }
    }
  ]
}

[
  {"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {"name": "analyze_text", "arguments": {"text": "hello"}}},
  {"jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": {"name": "translate", "arguments": {"text": "hello", "target": "zh"}}},
  {"jsonrpc": "2.0", "method": "notifications/initialized"}
]

{
  "jsonrpc": "2.0",
  "method": "notifications/progress",
  "params": {
    "progressToken": "task_123",
    "progress": 65,
    "total": 100,
    "message": "数据清洗中：已处理 12000/20000 条记录"
  }
}

// ❌ 旧版：双端点 HTTP+SSE
app.get('/sse', handleSSE);
app.post('/message', handleMessage);

// ✅ 新版：单端点 Streamable HTTP
app.post('/mcp', handleStreamableHTTP);
app.get('/mcp', handleSSEStream); // 可选：用于服务端主动推送

// 新增元数据发现端点
app.get('/.well-known/oauth-authorization-server', (req, res) => {
  res.json({ /* 授权服务器配置 */ });
});

// 新增动态客户端注册
app.post('/register', handleDynamicRegistration);

// 所有 MCP 端点添加认证中间件
app.post('/mcp', authenticateOAuth, handleMCP);

// ❌ 旧版：仅有基础描述
{ name: 'delete_file', description: 'Delete a file' }

// ✅ 新版：包含行为元数据
{
  name: 'delete_file',
  description: 'Delete a file',
  annotations: {
    title: 'Delete File',
    readOnlyHint: false,
    destructiveHint: true,
    idempotentHint: false,
    openWorldHint: false
  }
}

app.post('/mcp', (req, res) => {
  // 处理 Mcp-Session-Id
  const sessionId = req.headers['mcp-session-id'] || generateSessionId();
  res.setHeader('Mcp-Session-Id', sessionId);

  // 支持批处理
  if (Array.isArray(req.body)) {
    const results = req.body.map(msg => processMessage(msg, sessionId));
    return res.json(results.filter(Boolean));
  }
  // 单消息处理
  const result = processMessage(req.body, sessionId);
  result ? res.json(result) : res.status(202).end();
});