GitHub - Super-I-Tech/mcp_plexus: Secure, Multi-Tenant MCP Server Framework for Modern AI (original) (raw)

MCP Plexus: Secure, Multi-Tenant MCP Server Framework for Modern AI

Build powerful, scalable, and secure Model Context Protocol (MCP) applications with ease. MCP Plexus is a Python framework built on the robust jlowin/fastmcp (FastMCP 2.7) library, designed to empower developers to deploy multi-tenant MCP servers that seamlessly integrate with external services via OAuth 2.1 and manage API key access for tools.

Introduction

What is MCP Plexus?

MCP Plexus extends the capabilities of FastMCP 2.7, providing a structured environment for creating sophisticated, multi-tenant AI backend systems. It allows you to define distinct, isolated environments (tenants) that can expose customized sets of tools, resources, and prompts to Large Language Models (LLMs) and AI agents.

Key differentiating features include:

Why MCP Plexus?

As AI applications grow in complexity, the need to securely provide LLMs with relevant context and executable capabilities becomes paramount. MCP Plexus addresses this by:

Core Philosophy

Key Features

Architecture Overview

MCP Plexus sits as an ASGI application (typically run with Uvicorn) and acts as a sophisticated intermediary between MCP clients and the underlying FastMCP server instance.

Getting Started

Prerequisites

Installation / Setup

  1. Clone the Repository:
    git clone https://github.com/Super-I-Tech/mcp_plexus mcp-plexus
    cd mcp-plexus
  2. Create and Activate a Virtual Environment:
    On Windows
    On macOS/Linux
  3. Install Dependencies:
    pip install -r requirements.txt
  4. **Configure Environment Variables (.env file):**Create a .env file in the project root (mcp-plexus/.env). You can copy from .env.example if provided, or create it manually.
    Key Variables:
    • HOST_APP_REGISTRATION_SECRET: CRITICAL. A strong, unique secret that host applications must provide to register their users with Plexus (via the X-Host-App-Secret header). Change the default generated value for any production or shared deployment.
    • PLEXUS_ENCRYPTION_KEY: CRITICAL. A Fernet encryption key used for encrypting sensitive data like API keys and external OAuth tokens stored in the database. Generate one using:
      python -c "from mcp_plexus.utils import generate_fernet_key; print(generate_fernet_key())"
      Place the output in your .env. Keep this key secret and back it up. Losing it means losing access to encrypted data.
    • STORAGE_BACKEND: Set to "sqlite" (default) or "redis" for most persistent stores (e.g., Plexus user auth tokens, external OAuth provider configs).
      * IMPORTANT: MCP Session Management currently REQUIRES Redis and uses RedisPlexusSessionStore regardless of this setting. Future updates may enable SQLite for session storage.
    • SQLITE_DB_PATH: Path to the SQLite database file (e.g., ./mcp_plexus_data.sqlite3).
    • REDIS_HOST, REDIS_PORT, REDIS_DB, REDIS_PASSWORD (optional), REDIS_SSL (optional): Connection details for your Redis instance (REQUIRED FOR MCP SESSIONS; also used if STORAGE_BACKEND=redis for other data).
    • DEBUG_MODE: Set to True for development (more verbose logging, Uvicorn reload).
    • PLEXUS_FASTMCP_LOG_LEVEL: Logging level for the FastMCP component (e.g., DEBUG, INFO).
    • ADMIN_API_KEY: A secret API key for accessing administrative endpoints (e.g., managing tenants, external OAuth providers). Set a strong value.
      Example .env:

Uvicorn Development Server

DEV_SERVER_HOST=127.0.0.1
DEV_SERVER_PORT=8000
DEV_SERVER_LOG_LEVEL=info
DEV_SERVER_RELOAD=True

Application Settings

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 # see mcp_plexus/utils/generate_key.py 5. **Initialize Database (if using SQLite):**The SQLite database and tables are typically created automatically on first run if they don't exist.

Running the Server (Development)

Use the run_dev.py script:

This starts the Uvicorn server, usually on http://127.0.0.1:8000 (host/port can be configured via DEV_SERVER_HOST/DEV_SERVER_PORT in .env).

Using MCP Plexus (Developer Guide)

MCP Plexus acts as a framework or SDK for building your custom multi-tenant MCP server. Here's how you typically use it:

1. Defining Tenants

Tenants (entities) are the top-level organizational units.

2. Registering Host Application Users

To enable persistent storage of external OAuth tokens across sessions for a specific user of your main application (the "host application"), you first need to register that user identity with MCP Plexus for a given tenant.

3. Initializing MCP Sessions

MCP clients interact with a specific tenant's MCP endpoint.

4. Creating MCP Tools, Resources, & Prompts

You define tools, resources, and prompts in Python modules located within the mcp_plexus/tool_modules/ directory. These modules use the globally available PLEXUS_SERVER_INSTANCE (which is an instance of MCPPlexusServer) to register their components.

mcp_plexus/core/global_registry.py:

This instance is populated at server startup

PLEXUS_SERVER_INSTANCE: Optional[MCPPlexusServer] = None

Example Tool Module (mcp_plexus/tool_modules/my_custom_tools.py):

import logging from typing import Dict, Any, Optional, List import httpx # Required for @requires_auth example from fastmcp import Context as FastMCPBaseContext # For type hinting standard ctx from mcp_plexus.core.global_registry import PLEXUS_SERVER_INSTANCE from mcp_plexus.plexus_context import PlexusContext # For specific Plexus context features 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: # This check helps catch issues if tool modules are imported before PLEXUS_SERVER_INSTANCE is set 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"] # Tool only visible to these tenants ) async def get_greeting(ctx: FastMCPBaseContext) -> Dict[str, str]: plexus_ctx: PlexusContext = PlexusContext(ctx.fastmcp) # Create PlexusContext from base

entity_id = plexus_ctx.entity_id
user_id = plexus_ctx.persistent_user_id # Will be None for guest users
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}

Example: Tool requiring external GitHub OAuth

@PLEXUS_SERVER_INSTANCE.tool( name="get_my_github_repos", description="Fetches the user's GitHub repositories.", tool_sets=["developer_tools"], # Categorize tool allowed_tenant_ids=["mycompany"] ) @requires_auth(provider_name="github", scopes=["repo", "read:user"]) async def get_my_github_repos( ctx: FastMCPBaseContext, *, # This enforces subsequent arguments as keyword-only _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()

Example: Tool requiring an API key

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 # Injected by @requires_api_key ) -> 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.") # Use openweathermap_api_key to call the external weather API # Example: 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)"}

logger.info("my_custom_tools.py - Tools registered with PLEXUS_SERVER_INSTANCE.")

5. Securing Tools

a. External OAuth (@requires_auth)

Used when a tool needs to access a third-party service (e.g., GitHub, Google) on behalf of the user.

Flow:

  1. Admin Configures Provider: An administrator uses the CLI (or Admin API) to register the external OAuth provider details (client ID, secret, URLs, default scopes) for a specific tenant.
  2. Tool Decoration: Decorate your tool function with @requires_auth(provider_name="github", scopes=["repo", "read:user"]).
  3. Tool Call:
    • When an MCP client calls the tool, the decorator (via PlexusContext) checks for a valid, stored OAuth token for the current user (persistent if Plexus user, session if guest) and the requested provider/scopes.
    • Token Valid: The decorator injects an authenticated httpx.AsyncClient (named authenticated_client) into your tool's kwargs.
    • Token Invalid/Missing: The decorator raises a PlexusExternalAuthRequiredError. This is caught by FastMCP and converted into a structured MCP ToolError. The error payload includes an authorization_url.
  4. Client Handles Redirect: The MCP client application is responsible for redirecting the user to this authorization_url.
  5. User Authentication & Consent: The user authenticates with the external provider and grants consent.
  6. Callback to Plexus: The external provider redirects the user back to /{entity_id}/oauth/external_callback/{provider_name}.
  7. Token Exchange & Storage: MCP Plexus exchanges the authorization code for tokens, stores them securely (session and/or persistent user store), and displays a success page.
  8. Client Retries Tool: The user returns to the MCP client, which should then retry the original tool call. This time, PlexusContext will find the valid token.

graph TD subgraph Tool Developer A[Define Tool with @requires_auth] --> B{Tool Called by LLM/Client} end

subgraph MCP Plexus
    B --> C{PlexusContext: Token Check};
    C -- Valid Token --> D[Inject Auth_Client];
    D --> E[Execute Tool Logic];
    C -- Invalid/No Token --> F[Raise PlexusExternalAuthRequiredError];
    F --> G[Return MCP ToolError with auth_url];
end

subgraph MCP Client App
    G --> H[Redirect User to auth_url];
end

subgraph User & External Provider
    H --> I[User Authenticates & Consents at Ext. Provider];
    I --> J[Ext. Provider Redirects to Plexus /callback];
end

subgraph MCP Plexus
    J --> K{Plexus /callback Endpoint};
    K --> L[Exchange Auth Code for Tokens];
    L --> M[Store Tokens Session/Persistent];
    M --> N[Show Success Page to User];
end

subgraph MCP Client App
    N --> O[User Returns to Client App];
    O --> P[Client Retries Tool Call];
end

P --> C;
E --> Q[Return Tool Result to Client];

Loading

b. API Key Access (@requires_api_key)

Used when a tool needs a direct API key from the user to interact with an external service.

Flow:

  1. Tool Decoration: Decorate tool with @requires_api_key(provider_name="my_service", key_name_display="My Service API Key", instructions="Get your key from ...").
  2. Tool Call:
    • The decorator (via PlexusContext) checks if an API key for the current user (persistent if Plexus user) and provider is stored.
    • Key Found: The decrypted API key is injected into the tool's kwargs as {provider_name}_api_key (e.g., my_service_api_key).
    • Key Not Found: Raises PlexusApiKeyRequiredError. FastMCP converts this to a structured MCP ToolError instructing the client/user on how to submit the key.
  3. User Submits API Key: The user (via the MCP client or another interface) calls the POST /{entity_id}/plexus-services/api-keys endpoint.
    • Request Header: Authorization: Bearer <plexus_user_auth_token> (required for persistent storage).
    • Request Body (JSON):
      {
      "provider_name": "my_service",
      "api_key_value": "users_actual_api_key_value"
      }
  4. Key Storage: MCP Plexus encrypts and stores the API key associated with the user's persistent_user_id.
  5. Client Retries Tool: The tool call is retried, and the key is now found and injected.

graph TD subgraph Tool Developer A[Define Tool with @requires_api_key] --> B{Tool Called by LLM/Client} end

subgraph MCP Plexus
    B --> C{PlexusContext: API Key Check};
    C -- Key Found & Decrypted --> D[Inject API Key into Tool];
    D --> E[Execute Tool Logic];
    C -- Key Not Found --> F[Raise PlexusApiKeyRequiredError];
    F --> G[Return MCP ToolError with submission instructions];
end

subgraph User/MCP Client App
    G --> H[User/Client Submits API Key via /api-keys endpoint];
end

subgraph MCP Plexus
    H --> I{Plexus /api-keys Endpoint};
    I --> J[Encrypt & Store API Key];
end

subgraph User/MCP Client App
    J --> K[Client Retries Tool Call];
end

K --> C;
E --> L[Return Tool Result to Client];

Loading

Project Status & Roadmap

This section outlines the current implementation status and planned features.

Implemented

Planned / Work In Progress

Contributing

Details on contributing to MCP Plexus will be added soon. We welcome contributions for bug fixes, feature enhancements, and documentation improvements.

License

This project is licensed under the same licensed as FastMCP.