AI导哈

多租户认证框架

为MCP服务器提供可插拔的认证系统，支持多租户、多种认证方式和传输协议。

  • by prmichaelsen
  • Shell
  • 本地部署
  • 更新于
在 GitHub 查看 prmichaelsen/mcp-auth

使用说明

@prmichaelsen/mcp认证

MCP（模型上下文协议）服务器的身份验证和多租户框架。

概述

@prmichaelsen/mcp-auth 为MCP服务器提供可插拔的身份验证系统，实现：

  • 零修改：包装现有的MCP服务器，不更改代码
  • 多租户技术：具有单独资源令牌的多个用户
  • 身份不可知：支持JWT、环境变量或自定义身份验证方案
  • 运输无关：适用于stdio、HTTP和SSE传输
  • 类型安全：完全支持TypeScript
  • 可组合：用于速率限制、日志记录、超时、重试的中间件

两种模式

模式1：服务器包装（建议用于生产环境）

无需修改即可包装现有的MCP服务器：

import { wrapServer, JWTAuthProvider, APITokenResolver } from '@prmichaelsen/mcp-auth';
import { createServer } from '@myorg/my-mcp-server';

const wrapped = wrapServer({
  serverFactory: createServer,
  
  // Validates JWT from tenant manager
  authProvider: new JWTAuthProvider({
    jwtSecret: process.env.JWT_SECRET
  }),
  
  // Resolves tokens via tenant manager API (recommended)
  tokenResolver: new APITokenResolver({
    tenantManagerUrl: process.env.TENANT_MANAGER_URL,
    serviceToken: process.env.SERVICE_TOKEN
  }),
  
  resourceType: 'myapi',
  transport: { type: 'sse', port: 3000 }
});

await wrapped.start();

模式2：工具级身份验证

构建具有集成身份验证的新服务器：

import { AuthenticatedMCPServer, withAuth, EnvAuthProvider, SimpleTokenResolver } from '@prmichaelsen/mcp-auth';

const server = new AuthenticatedMCPServer({
  name: 'my-server',
  authProvider: new EnvAuthProvider(),
  tokenResolver: new SimpleTokenResolver({ tokenEnvVar: 'API_TOKEN' }),
  resourceType: 'myapi',
  transport: { type: 'stdio' }
});

server.registerTool('get_data', withAuth(async (args, accessToken, userId) => {
  const client = new MyAPIClient(accessToken);
  return client.getData(args);
}));

await server.start();

安装

# Core package
npm install @prmichaelsen/mcp-auth @modelcontextprotocol/sdk

# For JWT support
npm install jsonwebtoken

# For SSE/HTTP transports
npm install express cors

身份验证提供者

EnvAuthProvider（单用户）

对于本地开发和单用户场景：

import { EnvAuthProvider, SimpleTokenResolver } from '@prmichaelsen/mcp-auth';

const authProvider = new EnvAuthProvider({
  userIdEnvVar: 'MCP_USER_ID',
  defaultUserId: 'local-user'
});

const tokenResolver = new SimpleTokenResolver({
  tokenEnvVar: 'API_TOKEN'
});

JWTAuthProvider+APITokenResolver（多租户生产）⭐ 推荐

对于生产多租户部署：

import { JWTAuthProvider, APITokenResolver } from '@prmichaelsen/mcp-auth';

// Validates JWT tokens from tenant manager
const authProvider = new JWTAuthProvider({
  jwtSecret: process.env.JWT_SECRET,
  userIdClaim: 'sub'
});

// Resolves tokens via tenant manager API
const tokenResolver = new APITokenResolver({
  tenantManagerUrl: 'https://tenant-manager.example.com',
  serviceToken: process.env.SERVICE_TOKEN,
  cacheTokens: true, // Cache for performance
  cacheTtl: 300000 // 5 minutes
});

为什么基于API更好：

  • ✅ 自动令牌刷新（不需要新的JWT）
  • ✅ 立即撤销令牌
  • ✅ JWT中未公开的令牌
  • ✅ 小JWT大小（~200字节）
  • ✅ 集中式令牌管理

JWTAuthProvider+JWTTokenResolver（MVP/原型）

要快速设置JWT嵌入式令牌：

import { JWTAuthProvider, JWTTokenResolver } from '@prmichaelsen/mcp-auth';

const authProvider = new JWTAuthProvider({
  jwtSecret: process.env.JWT_SECRET,
  extractTokens: true // Extract tokens from JWT
});

const tokenResolver = new JWTTokenResolver({ authProvider });

权衡：

  • ✅ 零API调用（更快）
  • ✅ 更易于实施
  • ❌ 代币轮换需要新的JWT
  • ❌ JWT尺寸更大
  • ❌ JWT负载中暴露的令牌

JWTAuthProvider（静态服务器）⭐ 新

对于管理自己的数据并且只需要用户标识的服务器：

import { wrapServer, JWTAuthProvider } from '@prmichaelsen/mcp-auth';

const wrapped = wrapServer({
  serverFactory: (accessToken, userId) => {
    // accessToken will be empty string - use userId only
    return createMyStaticServer(userId);
  },
  
  authProvider: new JWTAuthProvider({
    jwtSecret: process.env.JWT_SECRET
  }),
  
  // No tokenResolver needed! ✨
  
  resourceType: 'my-service',
  transport: {
    type: 'sse',
    port: 3000,
    cors: true,
    corsOrigin: process.env.CORS_ORIGIN
  }
});

非常适合：

  • ✅ 多租户SaaS，拥有自己的数据库
  • ✅ 用户范围的服务
  • ✅ 没有外部API的内部工具
  • ✅ 静态数据管理服务器

优点：

  • ✅ 最简单的配置
  • ✅ 无外部凭据管理
  • ✅ 仅JWT验证（用户ID提取）
  • ✅ 通过临时实例实现完全的用户隔离

APITokenResolver（基于API）

对于通过租户管理器API解析令牌：

import { JWTAuthProvider, APITokenResolver } from '@prmichaelsen/mcp-auth';

const authProvider = new JWTAuthProvider({
  jwtSecret: process.env.JWT_SECRET
});

const tokenResolver = new APITokenResolver({
  tenantManagerUrl: 'https://tenant-manager.example.com',
  serviceToken: process.env.SERVICE_TOKEN,
  endpointPath: '/api/credentials/:userId/:resourceType'
});

自定义提供者

实现自己的身份验证逻辑：

import { AuthProvider, AuthResult, RequestContext } from '@prmichaelsen/mcp-auth';

class CustomAuthProvider implements AuthProvider {
  async authenticate(context: RequestContext): Promise<AuthResult> {
    const apiKey = context.headers?.['x-api-key'];
    
    if (!apiKey) {
      return { authenticated: false, error: 'No API key' };
    }
    
    // Your validation logic
    return {
      authenticated: true,
      userId: 'user-123'
    };
  }
}

中间件组成

import { compose, withAuth, withRateLimit, withLogging, withTimeout } from '@prmichaelsen/mcp-auth';

const getTool = compose(
  withLogging({ logArgs: true }),
  withRateLimit({ maxRequests: 100, windowMs: 60000 }),
  withTimeout(5000),
  withAuth(),
  async (args, accessToken, userId) => {
    // Your tool logic
  }
);

server.registerTool('get_data', getTool);

运输

Stdio（本地）

transport: { type: 'stdio' }

SSE（远程多租户）

transport: {
  type: 'sse',
  port: 3000,
  host: '0.0.0.0',
  basePath: '/mcp',
  cors: true,
  corsOrigin: 'https://your-app.example.com'  // REQUIRED when cors: true
}

创建的端点：

  • GET /mcp -服务器信息和可用端点
  • POST /mcp/message -MCP协议消息（需要JWT）
  • GET /mcp/health -健康检查端点

⚠️ CORS安全

启用CORS时，您 必须 指定 corsOrigin 有明确的起源：

// ✅ SECURE - Single origin
transport: {
  type: 'sse',
  cors: true,
  corsOrigin: 'https://app.example.com'
}

// ✅ SECURE - Multiple origins
transport: {
  type: 'sse',
  cors: true,
  corsOrigin: ['https://app1.example.com', 'https://app2.example.com']
}

// ❌ INSECURE - Wildcard blocked in production
transport: {
  type: 'sse',
  cors: true,
  corsOrigin: '*'  // ConfigurationError in production!
}

// ❌ INSECURE - Missing corsOrigin
transport: {
  type: 'sse',
  cors: true  // ConfigurationError: corsOrigin required!
}

安全要求：

  • corsOrigin 是 必需 当 cors: true
  • 通配符(*)是 仅 允许开发(NODE_ENV !== 'production')
  • 在生产中，通配符抛出 ConfigurationError 防止CSRF攻击
  • 使用特定的源来确保只有您的应用程序可以访问MCP服务器

HTTP（远程）

transport: {
  type: 'http',
  port: 3000,
  host: '0.0.0.0'
}

进度流

mcp-auth支持mcp进度通知，允许包装服务器在长时间运行的操作期间向客户端流式传输进度更新。

运作原理

当客户提供 progressToken 在请求中，mcp自动进行身份验证：

  1. 从请求中提取进度令牌
  2. 将其传递给包装好的MCP服务器
  3. 包装好的服务器可以将进度通知发送回客户端
  4. 每个用户的进度都是隔离的（多租户安全）

客户端使用情况

import { Client } from '@modelcontextprotocol/sdk/client/index.js';

const client = new Client({
  name: 'my-client',
  version: '1.0.0'
});

// Call tool with progress support
const result = await client.request({
  method: 'tools/call',
  params: {
    name: 'long_running_operation',
    arguments: { /* ... */ }
  }
}, {
  progressToken: 'operation-123',
  onprogress: (progress) => {
    console.log(`Progress: ${progress.progress}/${progress.total}`);
    console.log(`Message: ${progress.message}`);
  }
});

服务器端实现

封装的MCP服务器可以发送进度通知：

// In your MCP server tool handler
export async function handleLongOperation(
  args: any,
  extra?: { progressToken?: string | number }
): Promise<any> {
  const progressToken = extra?.progressToken;
  
  if (progressToken) {
    // Send progress notifications
    server.notification({
      method: 'notifications/progress',
      params: {
        progressToken,
        progress: 50,
        total: 100,
        message: 'Processing...'
      }
    });
  }
  
  // ... perform operation
}

特性

  • ✅ 自动通过：进度令牌自动转发到打包服务器
  • ✅ 多租户保险箱：按用户隔离进度通知
  • ✅ 向后兼容：使用或不使用进度标记（优雅降级）
  • ✅ 零配置：无需额外设置

监控

通过经过身份验证的端点监视活动进度流：

用户统计信息 （显示您自己的进度流）：

GET /mcp/progress/stats
Authorization: Bearer <jwt>

响应:

{
  "user": {
    "userId": "user123",
    "activeStreams": 2,
    "totalMessages": 1543,
    "totalBytes": 45231,
    "oldestStreamAge": 120000
  },
  "global": {
    "activeStreams": 15,
    "totalMessages": 8234,
    "userCount": 8
  },
  "timestamp": "2026-02-23T21:00:00.000Z"
}

所有指标 （详细流信息）：

GET /mcp/progress/metrics
Authorization: Bearer <jwt>

响应:

{
  "metrics": [
    {
      "userId": "user123",
      "progressToken": "op-456",
      "startTime": 1708725600000,
      "lastUpdate": 1708725650000,
      "duration": 50000,
      "messageCount": 125,
      "bytesTransferred": 45231,
      "averageMessageSize": 361.8,
      "messagesPerSecond": 2.5
    }
  ],
  "health": {
    "healthy": true,
    "issues": [],
    "warnings": []
  },
  "timestamp": "2026-02-23T21:00:00.000Z"
}

演出

  • 记忆：每个活动流约1KB
  • 网络：每个进度通知约100-500字节
  • 中央处理器：典型工作负载的开销<1%
  • 清理：每分钟自动删除过期流（>5分钟空闲）

备注

  • 进度流需要SSE或HTTP传输（stdio不可用）
  • 请求完成后，进度令牌会自动清理
  • 打包服务器必须实现进度通知支持才能发送更新
  • 健康检查检测到过时的流和高消息速率
  • 监控端点需要身份验证

MCP服务器合同

使您的MCP服务器与兼容 wrapServer()，导出工厂函数：

import { Server } from '@modelcontextprotocol/sdk/server/index.js';

export function createServer(accessToken: string, userId?: string): Server {
  const server = new Server({ name: 'my-server', version: '1.0.0' });
  const client = new MyAPIClient(accessToken);
  
  // Register your tool handlers...
  
  return server;
}

就是这样！服务器中不需要mcp身份验证导入。

建筑

三层部署

  1. 聊天平台 -使用JWT发送MCP请求
  2. 租户经理 -发布JWT，管理用户凭据
  3. MCP服务器实例 -使用mcp-auth包装mcp服务器

令牌解析方法

方法1：带有嵌入式令牌的JWT （推荐）

  • 租户管理器在JWT中包含资源令牌
  • 零外部呼叫
  • 最快的性能

方法2：基于API的解决方案

  • 租户管理器为令牌查找提供API
  • 更好地分离关注点
  • 更容易进行代币轮换

看 agent/token-resolution-approaches.md 了解详情。

文档

全面的架构文档 agent/:

例子

工作示例即将推出 examples/:

  • simple-stdio/ -单用户stdio服务器
  • wrapped-server/ -使用JWT进行服务器包装
  • tool-level-auth/ -工具级身份验证
  • jwt-multi-tenant/ -多租户JWT部署

API 参考

核心功能

  • wrapServer(config) -用身份验证包裹MCP服务器
  • withAuth(handler) -向基于函数的工具添加身份验证
  • compose(...middlewares) -编写中间件功能

类

  • AuthenticatedMCPServer -集成身份验证的MCP服务器
  • AuthenticatedTool -基于类的工具的包装器
  • BaseAuthProvider -身份验证提供者的基类

提供商

  • EnvAuthProvider -环境变量身份验证
  • SimpleTokenResolver -环境变量令牌解析
  • JWTAuthProvider -JWT验证与令牌提取
  • JWTTokenResolver -JWT嵌入式令牌解析
  • APITokenResolver -基于API的令牌解析

中间件

  • withAuth() -身份验证
  • withLogging() -请求/响应日志记录
  • withRateLimit() -每个用户的速率限制
  • withTimeout() -请求超时
  • withRetry() -失败时自动重试

示例项目

真实世界的项目使用 @prmichaelsen/mcp-auth:

@prmichaelsen/agentbase mcp服务器

具有JWT身份验证的多租户Instagram MCP服务器

  • 模式：服务器包装（动态模式）
  • 认证：基于JWT+API的令牌解析
  • 用例：外部API证书（Instagram）
  • 运输：基于HTTP的SSE

@prmichaelsen/记住mcp服务器

用于记住MCP的多租户MCP服务器，具有平台JWT身份验证

  • 模式：服务器包装（静态模式）
  • 认证：仅JWT（无tokenResolver）
  • 用例：自己数据库中的用户范围数据
  • 运输：基于HTTP的SSE

许可证

麻省理工学院

贡献

欢迎投稿！请在上打开问题或PR GitHub.

相关 MCP