麦普脉络

Name: 麦普脉络
Author: Super-I-Tech

作者：Super-I-Tech23 星标更新：2025-06-10

安装使用来源：spider==github.com

项目介绍

MCP Plexus：现代AI的多租户安全MCP服务器框架

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

简介

什么是MCP Plexus？

MCP Plexus扩展了FastMCP 2.7的功能，提供了一个结构化的环境来创建复杂的多租户AI后端系统。它允许您定义独立的环境（租户），这些环境可以向大型语言模型（LLMs）和AI代理暴露定制的工具、资源和提示集。

关键差异化特性包括：

强大的多租户支持： 在单个部署中托管多个客户或组织，每个都有自己的数据和工具访问权限。
简化外部服务集成： 安全地连接您的MCP工具到受OAuth 2.1保护的外部服务（如GitHub、Google API等），内置流程管理。
用户特定持久访问： 允许主机应用程序注册其用户与Plexus，允许持久存储外部OAuth令牌，减少重新认证摩擦。
工具的API密钥管理： 安全存储并注入需要直接密钥认证到外部服务的工具的API密钥。
标准化和可扩展性： 利用FastMCP的Python装饰器的全部功能来定义MCP组件，同时提供清晰的扩展结构。

为什么选择MCP Plexus？

随着AI应用复杂性的增加，安全地为LLMs提供相关上下文和执行能力变得至关重要。MCP Plexus通过以下方式解决了这一问题：

简化多租户部署： 减少管理多个隔离MCP环境的样板代码和复杂性。
标准化外部API访问： 提供一致且安全的机制，使工具能够与OAuth保护和API密钥门控的服务交互。
增强用户体验： 通过持久用户身份验证和令牌存储，最小化对外部服务的重复登录提示。
促进安全实践： 管理敏感凭证（OAuth令牌、API密钥）在核心工具逻辑之外。
利用FastMCP的力量： 基于经过验证的高性能FastM-CP 2.7库处理核心MCP协议及其友好的开发接口。

核心理念

安全第一： 设计时以OAuth 2.1和安全凭证管理为核心。
开发者体验： 目标是为定义租户、工具和安全集成提供直观的API。
可扩展性和隔离： 针对多租户架构设计，具有明确的关注点分离。
可扩展性： 提供一个基础，可以扩展自定义的身份验证提供商和服务。

关键特性

多租户：
- 支持通过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)装饰器。
- 如果尚未存储，通过结构化的MCP错误提示用户提交其API密钥。
- 使用PLEXUS_ENCRYPTION_KEY加密持久存储API密钥，并将其持久存储到persistent_user_id。
- 在运行时将解密的API密钥注入工具函数。
- 用户提交API密钥的端点（/{entity_id}/plexus-services/api-keys）。
标准MCP服务器功能（通过FastMCP 2.7）：
- 使用FastMCP的Python装饰器和类（例如PLEXUS_SERVER_INSTANCE.tool()）完全支持定义MCP工具、资源和提示。
- 原生流式HTTP传输用于MCP通信。
- 工具/资源/提示内的PlexusContext（扩展fastmcp.Context）访问日志、会话数据以及认证客户端或API密钥。
可配置存储：
- 默认使用SQLite进行Plexus用户身份验证令牌、用户特定的外部OAuth令牌、API密钥、外部OAuth提供商配置、租户配置和用户提交的API密钥的持久存储。
- 当前需要Redis进行MCP会话管理（计划增强SQLite会话存储）。
（正在进行的工作）内部OAuth 2.1提供商：
- MCP Plexus作为其自身的OAuth 2.1授权服务器的基础元素。
- 包括基本模型和存根端点。完整的实现（动态客户端注册、内部客户端的用户同意、完整的令牌管理）是未来的目标。

架构概述

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

FastAPI： 用于主要的Web应用程序路由，处理HTTP请求，并提供管理API端点。
FastMCP（jlowin/fastmcp）： 使用共享实例的FastMCP作为处理MCP协议消息（工具、资源、提示）的核心引擎，通过流式HTTP。
Plexus核心：
- 租户管理： 从URL路径（/{entity_id}/...）识别租户，并确保尊重租户配置（正在进行的完整配置工作，当前数据库支持验证）。
- 会话管理（PlexusSessionManager）： 管理会话数据（例如Mcp-Session-Id），当前使用Redis进行会话状态管理。它将MCP会话与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提供商配置和租户元数据。

快速开始

先决条件

Python 3.10+
Redis（目前用于MCP会话管理）
访问命令行/终端。

安装/设置

克隆仓库：

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

创建并激活虚拟环境：

python -m venv venv

在Windows上

.\venv\Scripts\activate

在macOS/Linux上

source venv/bin/activate

安装依赖项：
```
pip install -r requirements.txt
```
配置环境变量（.env文件）： 在项目根目录（mcp-plexus/.env）创建一个.env文件。如果您提供了.env.example，可以从那里复制，或者手动创建。

关键变量：
- HOST_APP_REGISTRATION_SECRET：关键。 主机应用程序必须提供的强唯一密钥，以注册其用户与Plexus（通过X-Host-App-Secret头）。更改任何生产或共享部署的默认生成值。
- PLEXUS_ENCRYPTION_KEY：关键。 用于加密存储在数据库中的敏感数据（如API密钥和外部OAuth令牌）的Fernet加密密钥。使用以下命令生成：
```
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提供商）的秘密API密钥。设置一个强值。
示例.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
```
初始化数据库（如果使用SQLite）： 如果不存在，通常会在第一次运行时自动创建SQLite数据库和表。

运行服务器（开发）

使用run_dev.py脚本：

python run_dev.py

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

使用MCP Plexus（开发人员指南）

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的一个实例）来注册其组件。

mcp_plexus/core/global_registry.py：

# 此实例在服务器启动时填充
PLEXUS_SERVER_INSTANCE: Optional[MCPPlexusServer] = None

示例工具模块（mcp_plexus/tool_modules/my_custom_tools.py）：

import logging
from typing import Dict, Any, Optional, List
import httpx # 用于@requires_auth示例
from fastmcp import Context as FastMCPBaseContext # 用于标准ctx的类型提示
from mcp_plexus.core.global_registry import PLEXUS_SERVER_INSTANCE
from mcp_plexus.plexus_context import PlexusContext # 用于特定Plexus上下文功能
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:
    # 此检查有助于捕获在导入`my_custom_tools.py`之前未设置`PLEXUS_SERVER_INSTANCE`的问题
    raise RuntimeError("PLEXUS_SERVER_INSTANCE未初始化。")

@PLEXUS_SERVER_INSTANCE.tool(
    name="get_tenant_specific_greeting",
    description="返回特定于当前租户的问候。",
    allowed_tenant_ids=["mycompany", "another_tenant"] # 工具仅对这些租户可见
)
async def get_greeting(ctx: FastMCPBaseContext) -> Dict[str, str]:
    plexus_ctx: PlexusContext = PlexusContext(ctx.fastmcp) # 从基础创建PlexusContext
    
    entity_id = plexus_ctx.entity_id
    user_id = plexus_ctx.persistent_user_id # 对于访客用户将是None
    session_val = await plexus_ctx.get_session_value("my_key")

    greeting = f"来自实体'{entity_id}'的问候！"
    if user_id:
        greeting += f" 认证用户：{user_id}。"
    if session_val:
        greeting += f" 'my_key'的会话值：{session_val}。"
        
    return {"greeting": greeting}

# 示例：需要外部GitHub OAuth的工具
@PLEXUS_SERVER_INSTANCE.tool(
    name="get_my_github_repos",
    description="获取用户的GitHub存储库。",
    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"为用户