MCP Plexus

🚀 MCP Plexus：适用于现代人工智能的安全、多租户MCP服务器框架

轻松构建强大、可扩展且安全的模型上下文协议（MCP）应用程序。MCP Plexus是一个基于强大的jlowin/fastmcp（FastMCP 2.7）库构建的Python框架，旨在赋能开发者部署多租户MCP服务器，该服务器可通过OAuth 2.1与外部服务无缝集成，并管理工具的API密钥访问。

🚀 快速开始

前提条件

• Python 3.10及以上版本
• Redis（目前MCP会话管理需要）
• 可访问命令行/终端

安装与设置

1. 克隆仓库：

git clone https://github.com/Super-I-Tech/mcp_plexus mcp-plexus
cd mcp-plexus

2. 创建并激活虚拟环境：

python -m venv venv

在Windows系统上：

.\venv\Scripts\activate

在macOS/Linux系统上：

source venv/bin/activate

3. 安装依赖项：

pip install -r requirements.txt

4. 配置环境变量（.env文件）： 在项目根目录（mcp-plexus/.env）中创建一个.env文件。如果有.env.example文件，可以从中复制内容，也可以手动创建。

关键变量：

• HOST_APP_REGISTRATION_SECRET：至关重要。这是一个强大且唯一的密钥，主机应用程序必须提供此密钥才能向Plexus注册其用户（通过X-Host-App-Secret请求头）。对于任何生产环境或共享部署，请更改默认生成的值。
• PLEXUS_ENCRYPTION_KEY：至关重要。这是一个Fernet加密密钥，用于加密存储在数据库中的敏感数据，如API密钥和外部OAuth令牌。可使用以下命令生成：

python -c "from mcp_plexus.utils import generate_fernet_key; print(generate_fernet_key())"

将输出结果放入.env文件中。请妥善保管此密钥并进行备份。丢失该密钥意味着无法访问加密数据。

• STORAGE_BACKEND：设置为"sqlite"（默认值）或"redis"，用于大多数持久存储（如Plexus用户认证令牌、外部OAuth提供者配置等）。
  • 重要提示：目前MCP会话管理需要Redis，并且无论此设置如何，都使用RedisPlexusSessionStore。未来更新可能会支持使用SQLite进行会话存储。
• SQLITE_DB_PATH：SQLite数据库文件的路径（例如./mcp_plexus_data.sqlite3）。
• REDIS_HOST、REDIS_PORT、REDIS_DB、REDIS_PASSWORD（可选）、REDIS_SSL（可选）：Redis实例的连接详细信息（MCP会话需要；如果STORAGE_BACKEND=redis，其他数据也会使用）。
• DEBUG_MODE：设置为True以开启开发模式（更详细的日志记录，Uvicorn自动重新加载）。
• PLEXUS_FASTMCP_LOG_LEVEL：FastMCP组件的日志记录级别（例如DEBUG、INFO）。
• ADMIN_API_KEY：用于访问管理端点的密钥（例如管理租户、外部OAuth提供者）。请设置一个强密钥。

示例.env文件：

# Uvicorn开发服务器
DEV_SERVER_HOST=127.0.0.1
DEV_SERVER_PORT=8000
DEV_SERVER_LOG_LEVEL=info
DEV_SERVER_RELOAD=True

# 应用程序设置
APP_NAME=MCP Plexus Server
DEBUG_MODE=True

REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_DB=0
# REDIS_PASSWORD=
# REDIS_SSL=False

PLEXUS_FASTMCP_LOG_LEVEL="DEBUG"

STORAGE_BACKEND=sqlite
SQLITE_DB_PATH=./mcp_plexus_data.sqlite3

HOST_APP_REGISTRATION_SECRET=host_app_secre

ADMIN_API_KEY=your_super_secret_admin_api_key_here_12345

PLEXUS_CLI_API_BASE_URL=http://127.0.0.1:8080

PLEXUS_ENCRYPTION_KEY=your_generated_fernet_key_here # 请参考mcp_plexus/utils/generate_key.py

5. 初始化数据库（如果使用SQLite）： 如果SQLite数据库和表不存在，通常在首次运行时会自动创建。

运行服务器（开发环境）

使用run_dev.py脚本：

python run_dev.py

这将启动Uvicorn服务器，通常运行在http://127.0.0.1:8000（主机和端口可以通过.env文件中的DEV_SERVER_HOST和DEV_SERVER_PORT进行配置）。

✨ 主要特性

• 多租户支持：
  • 通过URL路径（例如/{entity_id}/mcp/）识别多个隔离的租户。
  • 支持租户特定的MCP会话管理（目前默认使用Redis）。
  • （计划中）支持租户特定的工具可见性和提供者设置配置。
• Plexus用户认证：
  • 主机应用程序可以通过安全端点（/{entity_id}/plexus-auth/register-user）向MCP Plexus注册其用户。
  • 获取plexus_user_auth_token，该令牌将主机应用程序的用户与Plexus内的持久身份关联起来。
  • 支持持久存储用户特定的外部OAuth令牌和API密钥。
• 工具的外部OAuth 2.1流程简化：
  • MCP工具使用@requires_auth(provider_name, scopes)装饰器触发与外部提供者（如GitHub）的OAuth 2.1授权码授予流程（带有PKCE）。
  • 管理OAuth回调、令牌交换和安全的令牌存储（对于访客用户，令牌与会话绑定；对于已认证的Plexus用户，令牌持久存储在persistent_user_id下）。
  • 工具通过PlexusContext接收经过身份验证的httpx.AsyncClient，用于与外部提供者进行交互。
  • 提供管理租户特定外部OAuth提供者配置的管理CLI/API（例如GitHub的客户端ID/密钥）。
• 工具的API密钥管理：
  • MCP工具使用@requires_api_key(provider_name, key_name_display, instructions)装饰器。
  • 如果用户尚未存储API密钥，会通过结构化的MCP错误提示用户提交。
  • 安全地将API密钥（使用PLEXUS_ENCRYPTION_KEY进行静态加密）持久存储在persistent_user_id下。
  • 在运行时将解密后的API密钥注入工具函数。
  • 提供端点（/{entity_id}/plexus-services/api-keys）供用户提交API密钥。
• 标准MCP服务器功能（通过FastMCP 2.7）：
  • 完全支持使用FastMCP的Python装饰器和类（例如PLEXUS_SERVER_INSTANCE.tool()）定义MCP工具、资源和提示。
  • 支持原生可流式传输的HTTP传输进行MCP通信。
  • 在工具/资源/提示中可以访问PlexusContext（扩展自fastmcp.Context），用于日志记录、会话数据以及访问经过身份验证的客户端或API密钥。
• 可配置的存储：
  • 默认使用SQLite持久存储Plexus用户认证令牌、用户特定的外部OAuth令牌、API密钥、外部OAuth提供者配置、租户配置和用户提交的API密钥。
  • 目前MCP会话管理需要Redis（计划增强SQLite会话存储功能）。
• （开发中）内部OAuth 2.1提供者：
  • MCP Plexus作为自己的OAuth 2.1授权服务器的基础元素。
  • 包括基本模型和存根端点。完整实现（动态客户端注册、用户对内部客户端的同意、完整的令牌管理）是未来的目标。

💻 使用示例

基础用法

以下是一个简单的示例，展示如何使用MCP Plexus定义一个工具：

import logging
from typing import Dict, Any, Optional, List
import httpx
from fastmcp import Context as FastMCPBaseContext
from mcp_plexus.core.global_registry import PLEXUS_SERVER_INSTANCE
from mcp_plexus.plexus_context import PlexusContext
from mcp_plexus.oauth.decorators import requires_auth
from mcp_plexus.services.decorators import requires_api_key

logger = logging.getLogger(__name__)

if PLEXUS_SERVER_INSTANCE is None:
    raise RuntimeError("PLEXUS_SERVER_INSTANCE not initialized when my_custom_tools.py was imported.")

@PLEXUS_SERVER_INSTANCE.tool(
    name="get_tenant_specific_greeting",
    description="Returns a greeting specific to the current tenant.",
    allowed_tenant_ids=["mycompany", "another_tenant"]
)
async def get_greeting(ctx: FastMCPBaseContext) -> Dict[str, str]:
    plexus_ctx: PlexusContext = PlexusContext(ctx.fastmcp)
    entity_id = plexus_ctx.entity_id
    user_id = plexus_ctx.persistent_user_id
    session_val = await plexus_ctx.get_session_value("my_key")

    greeting = f"Hello from entity '{entity_id}'!"
    if user_id:
        greeting += f" Authenticated user: {user_id}."
    if session_val:
        greeting += f" Session value for 'my_key': {session_val}."

    return {"greeting": greeting}

高级用法

以下示例展示了如何使用@requires_auth和@requires_api_key装饰器：

# 示例：需要外部GitHub OAuth的工具
@PLEXUS_SERVER_INSTANCE.tool(
    name="get_my_github_repos",
    description="Fetches the user's GitHub repositories.",
    tool_sets=["developer_tools"],
    allowed_tenant_ids=["mycompany"]
)
@requires_auth(provider_name="github", scopes=["repo", "read:user"])
async def get_my_github_repos(
    ctx: FastMCPBaseContext,
    *,
    _authenticated_client: httpx.AsyncClient
) -> List[Dict[str, Any]]:
    plexus_ctx = PlexusContext(ctx.fastmcp)
    logger.info(f"Fetching GitHub repos for user '{plexus_ctx.persistent_user_id}' in entity '{plexus_ctx.entity_id}'")
    response = await _authenticated_client.get("https://api.github.com/user/repos")
    response.raise_for_status()
    return response.json()

# 示例：需要API密钥的工具
WEATHER_API_PROVIDER_NAME = "openweathermap"
@PLEXUS_SERVER_INSTANCE.tool(
    name="get_weather_forecast",
    description=f"Gets weather forecast using {WEATHER_API_PROVIDER_NAME}.",
    allowed_tenant_ids=["mycompany", "another_tenant"]
)
@requires_api_key(
    provider_name=WEATHER_API_PROVIDER_NAME,
    key_name_display="OpenWeatherMap API Key",
    instructions="Please provide an API key for OpenWeatherMap."
)
async def get_weather(
    ctx: FastMCPBaseContext,
    city: str,
    *,
    openweathermap_api_key: str
) -> Dict[str, Any]:
    plexus_ctx = PlexusContext(ctx.fastmcp)
    logger.info(f"Fetching weather for {city} for user '{plexus_ctx.persistent_user_id}' in entity '{plexus_ctx.entity_id}' using provided API key.")
    # 使用openweathermap_api_key调用外部天气API
    # 示例：weather_data = await httpx.get(f"api.openweathermap.org/data/2.5/weather?q={city}&appid={openweathermap_api_key}")
    return {"city": city, "temperature": "20C", "condition": "Sunny (mock data)"}

📚 详细文档

MCP Plexus作为一个框架或SDK，用于构建自定义的多租户MCP服务器。以下是通常的使用方法：

1. 定义租户

租户（实体）是顶级组织单元。

• 管理：目前可以通过管理CLI命令（或者如果构建了管理界面，也可以通过直接的API调用）来管理租户。
  • 示例CLI命令：

plexus admin tenant create --entity-id "mycompany" --name "My Company Inc."

• 访问：每个租户都有自己的MCP端点：http://<server>/{entity_id}/mcp/

2. 注册主机应用程序用户

为了实现为你的主应用程序（“主机应用程序”）的特定用户跨会话持久存储外部OAuth令牌，你首先需要为给定的租户将该用户身份注册到MCP Plexus。

• 端点：POST /{entity_id}/plexus-auth/register-user
• 请求头：X-Host-App-Secret: <your_HOST_APP_REGISTRATION_SECRET_value>
• 请求体（JSON）：

{
    "user_id_from_host_app": "unique_stable_user_id_from_your_main_app"
}

• 响应体（JSON）：

{
    "plexus_user_auth_token": "a_long_secure_token_string_for_this_user",
    "persistent_user_id": "unique_stable_user_id_from_your_main_app",
    "message": "User token processed successfully."
}

你的主机应用程序应该安全地存储返回的plexus_user_auth_token，并使用它为该用户对后续的MCP请求进行身份验证。

3. 初始化MCP会话

MCP客户端与特定租户的MCP端点进行交互。

• 端点：/{entity_id}/mcp/
• 身份验证：
  • 对于已认证的Plexus用户：在所有MCP请求中，包括初始的initialize调用，通过Authorization: Bearer <token>HTTP头包含在步骤2中获得的plexus_user_auth_token。
• 标准MCP初始化请求：

{
    "jsonrpc": "2.0",
    "method": "initialize",
    "id": "client-init-123",
    "params": {
        "protocolVersion": "2025-03-26",
        "capabilities": {},
        "clientInfo": {
            "name": "MyHostApplicationClientName",
            "version": "1.0.0"
        }
    }
}

服务器将以Mcp-Session-Id头进行响应，客户端必须在该会话的后续请求中包含此头。

4. 创建MCP工具、资源和提示

你可以在mcp_plexus/tool_modules/目录下的Python模块中定义工具、资源和提示。这些模块使用全局可用的PLEXUS_SERVER_INSTANCE（MCPPlexusServer的一个实例）来注册它们的组件。

5. 保护工具

a. 外部OAuth (@requires_auth)

当工具需要代表用户访问第三方服务（如GitHub、Google）时使用。

流程：

1. 管理员配置提供者：管理员使用CLI（或管理API）为特定租户注册外部OAuth提供者的详细信息（客户端ID、密钥、URL、默认范围）。

• 示例CLI命令：

plexus admin ext-oauth create --entity-id "mycompany" --provider-name "github" --client-id "YOUR_GITHUB_CLIENT_ID" --client-secret "YOUR_GITHUB_CLIENT_SECRET" --authorization-url "https://github.com/login/oauth/authorize" --token-url "https://github.com/login/oauth/access_token" --scopes "repo,user:email" --userinfo-url "https://api.github.com/user"

2. 工具装饰：使用@requires_auth(provider_name="github", scopes=["repo", "read:user"])装饰你的工具函数。
3. 工具调用：

• 当MCP客户端调用工具时，装饰器（通过PlexusContext）会检查当前用户（如果是Plexus用户则为持久存储，如果是访客则为会话存储）是否有针对请求的提供者和范围的有效、存储的OAuth令牌。
• 令牌有效：装饰器会将经过身份验证的httpx.AsyncClient（名为authenticated_client）注入到你的工具的关键字参数中。
• 令牌无效/缺失：装饰器会抛出PlexusExternalAuthRequiredError。FastMCP会捕获此错误并将其转换为结构化的MCP ToolError。错误负载中包含一个authorization_url。

4. 客户端处理重定向：MCP客户端应用程序负责将用户重定向到这个authorization_url。
5. 用户身份验证和同意：用户使用外部提供者进行身份验证并授予同意。
6. 回调到Plexus：外部提供者将用户重定向回/{entity_id}/oauth/external_callback/{provider_name}。
7. 令牌交换和存储：MCP Plexus将授权码交换为令牌，安全地存储它们（会话存储和/或持久用户存储），并显示一个成功页面。
8. 客户端重试工具调用：用户返回MCP客户端，客户端应该重试原始的工具调用。这一次，PlexusContext将找到有效的令牌。

b. API密钥访问 (@requires_api_key)

当工具需要用户提供直接的API密钥才能与外部服务交互时使用。

流程：

1. 工具装饰：使用@requires_api_key(provider_name="my_service", key_name_display="My Service API Key", instructions="Get your key from ...")装饰工具。
2. 工具调用：

• 装饰器（通过PlexusContext）检查当前用户（如果是Plexus用户则为持久存储）是否有针对该提供者的API密钥。
• 密钥找到：解密后的API密钥将作为{provider_name}_api_key（例如my_service_api_key）注入到工具的关键字参数中。
• 密钥未找到：抛出PlexusApiKeyRequiredError。FastMCP会将此错误转换为结构化的MCP ToolError，指示客户端/用户如何提交密钥。

3. 用户提交API密钥：用户（通过MCP客户端或其他界面）调用POST /{entity_id}/plexus-services/api-keys端点。

• 请求头：Authorization: Bearer <plexus_user_auth_token>（持久存储需要）。
• 请求体（JSON）：

{
    "provider_name": "my_service",
    "api_key_value": "users_actual_api_key_value"
}

4. 密钥存储：MCP Plexus对API密钥进行加密并将其与用户的persistent_user_id关联存储。
5. 客户端重试工具调用：客户端重试工具调用，此时密钥将被找到并注入。

🔧 技术细节

MCP Plexus作为一个ASGI应用程序（通常使用Uvicorn运行），并作为MCP客户端和底层FastMCP服务器实例之间的复杂中介。

• FastAPI：用于主Web应用程序路由、处理HTTP请求和提供管理API端点。
• FastMCP (jlowin/fastmcp)：使用FastMCP的共享实例作为核心引擎，通过可流式传输的HTTP处理MCP协议消息（工具、资源、提示）。
• Plexus核心：
  • 租户管理：从URL路径（/{entity_id}/...）识别租户，并确保遵守租户配置（完整配置仍在开发中，目前基于数据库进行验证）。
  • 会话管理 (PlexusSessionManager)：管理MCP会话数据（例如Mcp-Session-Id），目前使用Redis存储会话状态。它将会话与entity_id和（如果可用）persistent_user_id关联起来。
  • Plexus用户认证：处理主机应用程序用户的注册，并管理plexus_user_auth_token值，以链接到persistent_user_id。
  • OAuth流程编排：管理工具访问外部服务所需的OAuth 2.1授权码授予流程，包括安全的令牌存储和刷新（刷新功能正在开发中）。
  • API密钥服务：管理用户提供的工具API密钥的安全存储和检索。
  • PlexusContext：注入到MCP工具中的增强上下文对象，提供对entity_id、persistent_user_id、会话数据的访问，以及获取经过身份验证的HTTP客户端或API密钥的辅助方法。
• 存储：
  • Redis（会话必需）：目前用于临时的MCP会话数据。
  • SQLite（默认用于持久数据）：用于存储Plexus用户认证令牌、用户特定的外部OAuth令牌、API密钥、外部OAuth提供者配置和租户元数据。

📄 许可证

本项目遵循与FastMCP相同的许可证。

项目状态与路线图

已实现功能

• [x] 核心多租户支持：通过URL（/{entity_id}/mcp/）进行基本的租户识别，并基于数据库进行验证。
• [x] FastMCP 2.7集成：使用jlowin/fastmcp处理核心MCP协议（可流式传输的HTTP）、工具/资源/提示定义。
• [x] 会话管理：基于Redis的MCP会话存储（PlexusSessionManager，SessionData）。
• [x] Plexus用户认证：主机应用程序用户注册（/{entity_id}/plexus-auth/register-user）以获取plexus_user_auth_token，用于链接到persistent_user_id。MCP请求通过Bearer令牌或URL令牌路径进行身份验证。
• [x] 外部OAuth 2.1流程简化：
  • 工具使用@requires_auth装饰器。
  • 处理OAuth回调（/{entity_id}/oauth/external_callback/{provider_name}）。
  • 为已认证的Plexus用户持久存储外部OAuth令牌（默认使用SQLite）。
  • 为访客用户提供会话范围的外部OAuth令牌存储。
  • 管理每个租户的外部OAuth提供者配置的管理API/CLI。
• [x] 工具的API密钥管理：
  • @requires_api_key装饰器。
  • 安全（加密）持久存储用户提交的API密钥（默认使用SQLite），并与persistent_user_id关联。
  • 提供API密钥提交端点（/{entity_id}/plexus-services/api-keys）。
• [x] PlexusContext：提供entity_id、persistent_user_id、mcp_session_id以及用于OAuth/API密钥访问的方法。
• [x] 管理端点和基本CLI：用于管理租户和外部OAuth提供者配置。
• [x] SQLite持久存储：作为用户认证令牌、外部令牌/API密钥、提供者配置、租户数据的默认后端。
• [x] 工具作用域和集合：基于allowed_tenant_ids实现基本的工具可见性，并通过tools/list参数按tool_sets进行过滤。

计划中/开发中的功能

• [ ] 完整的内部OAuth 2.1提供者：
  • [ ] 为内部OAuth客户端实现动态客户端注册。
  • [ ] 为内部OAuth客户端提供面向用户的同意屏幕。
  • [ ] 实现完整的令牌撤销和自省端点。
  • [ ] 提供更强大的安全性和授权类型支持。
• [ ] SQLite会话存储：实现SQLitePlexusSessionStore，使Redis完全可选。
• [ ] 细粒度的存储后端配置：允许为不同的数据类型（会话、持久令牌等）选择不同的后端（Redis/SQLite）。
• [ ] Plexus用户认证令牌生命周期管理：为plexus_user_auth_token实现强大的过期和撤销机制。
• [ ] 全面的租户配置：允许租户自定义其环境的更多方面（例如默认工具集、适用的UI主题）。
• [ ] 高级工具作用域：实现更复杂的工具可见性和访问控制规则。
• [ ] 增强的CLI：提供更多命令，提高管理和用户任务的可用性。
• [ ] 所有存储后端的完全异步支持：确保所有数据库操作都完全异步（如果尚未实现）。
• [ ] 更强大的错误处理和可观测性：在所有API中提供标准化的错误响应，改进结构化日志记录，并可能提供指标。
• [ ] 生产部署指南：提供关于Docker、反向代理、扩展的文档。
• [ ] 全面的测试套件：扩展所有功能的单元、集成和端到端测试。
• [ ] 详细的用户和开发者文档：将文档扩展到本README之外，形成结构化的文档网站。

贡献

关于如何为MCP Plexus做出贡献的详细信息将很快添加。我们欢迎对错误修复、功能增强和文档改进的贡献。