UnifiedNexus ⚡

One function. Two interfaces. Enterprise AI-agent tool serving for FastAPI + Model Context Protocol (MCP).

UnifiedNexus is a lightweight, secure, and extensible Python framework that bridges the gap between traditional REST APIs and AI-agent tools. Build customer-facing backends where your web frontend and autonomous AI agents share the exact same logic, schemas, and execution paths — defined once, with fail-closed security and trusted boundaries.

🎯 The Problem: Interface Drift & Insecure Tool Exposure

In traditional agentic systems, developers create a FastAPI route, then manually re-wrap it as an MCP tool. Over time, schemas drift, authentication gets bypassed, and internal database handles leak into LLM prompts.

❌ Traditional Duplication (Vulnerable & Drifting)
├── routes/orders.py     ← HTTP logic with FastAPI Depends(get_current_user)
├── mcp/orders.py        ← Copy-pasted MCP tool (bypasses auth, leaks DB sessions)
└── models/orders.py     ← Schema drift between OpenAPI and LLM json-schema

✅ UnifiedNexus (Single Source of Truth, Secure Boundaries)
└── main.py              ← One function. One decorator. Dual interfaces. Fail-closed security.

📦 Installation

pip install unified-nexus

🚀 Quickstart

from unified_nexus import UnifiedNexus, ToolSafetyLevel
from pydantic import BaseModel, Field

nexus = UnifiedNexus("CustomerAPI", version="0.2.0")

class UserRequest(BaseModel):
    user_id: int = Field(..., description="The unique ID of the user")

@nexus.universal_tool(path="/user-info", tool_type=ToolSafetyLevel.READ_ONLY, idempotent=True)
def get_user(req: UserRequest):
    """
    Fetches user status and clearance level.
    Docstrings and Pydantic schemas are auto-published to Swagger UI and MCP tools.
    """
    return {"id": req.user_id, "status": "Active", "tier": "Gold"}

# Build unified ASGI app
app = nexus.finalize()

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8000)

Run and inspect both interfaces simultaneously:

• 🌐 REST API & Swagger Docs: http://localhost:8000/docs
• 🤖 MCP SSE Transport: http://localhost:8000/mcp
• 🔍 MCP Inspector: npx -y @modelcontextprotocol/inspector http://localhost:8000/mcp

🛡️ Enterprise Security Model

UnifiedNexus treats REST and MCP as distinct transport boundaries. REST authentication does not automatically secure MCP tools, and MCP tools must never receive credentials as LLM arguments.

flowchart TD
    subgraph REST_Client["Web / Mobile Clients"]
        HTTP_Req["HTTP Request + Bearer Token"]
    end

    subgraph MCP_Client["AI Agents (Claude / Cursor / LangGraph)"]
        MCP_Req["MCP JSON-RPC Tool Call"]
    end

    subgraph Nexus["UnifiedNexus Boundary"]
        REST_Auth["FastAPI rest_auth / Depends"]
        MCP_Auth["FastMCP AuthProvider / Roles / Scopes"]
        Validation["Security Parameter Validator"]
        Execution["Execution Engine + Plugins"]
    end

    HTTP_Req --> REST_Auth --> Execution
    MCP_Req --> MCP_Auth --> Validation --> Execution

1. Fail-Closed Parameter Validation

• No FastAPI Depends() in MCP: Attempting to expose an endpoint containing Depends(...) or Security(...) with expose="both" or expose="mcp" immediately raises SecurityConfigurationError. FastAPI dependencies require HTTP headers and must not be blindly invoked in MCP.
• No Credential Arguments in LLM Schemas: Parameters named password, token, secret, or api_key are rejected from MCP registration, preventing LLMs from spoofing caller credentials.
• Router Security Isolation: APIRouters with router-level security dependencies (router.dependencies) cannot be converted to MCP without explicit MCP authorization policies.

2. Dual Authentication Configuration

from fastapi import Depends, Header, HTTPException
from unified_nexus import UnifiedNexus

def verify_jwt(authorization: str = Header(None)):
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Missing Bearer Token")
    return True

# Separate REST and MCP authentication mechanisms
nexus = UnifiedNexus(
    "SecureService",
    rest_auth=Depends(verify_jwt),  # Protects REST endpoints
    mcp_auth=my_mcp_token_verifier, # Protects MCP endpoint
)

🧩 Extensible Plugin Architecture

Plugins run identically across both REST and MCP requests, providing unified auditing, rate limiting, and request tracing.

from unified_nexus import NexusPlugin, NexusContext

class ComplianceAuditPlugin(NexusPlugin):
    async def startup(self, nexus):
        print("🚀 Compliance monitoring active.")

    async def before_execute(self, context: NexusContext, args: dict):
        print(f"[{context.request_id}] {context.transport.upper()} call to '{context.tool_name}'")

    async def after_execute(self, context: NexusContext, result: dict):
        print(f"[{context.request_id}] Executed in {context.duration_ms:.2f}ms")
        return result

    async def on_error(self, context: NexusContext, error: Exception):
        print(f"[{context.request_id}] Error in '{context.tool_name}': {error}")

nexus.add_plugin(ComplianceAuditPlugin())

NexusContext Attributes

🤖 AI-Agent Safety & Operational Invariants

⚙️ Exposure Control: App & Tool Levels

1. App-Level Exposure

nexus = UnifiedNexus("App", expose="both")  # REST + MCP (default)
nexus = UnifiedNexus("App", expose="api")   # REST Only (MCP disabled)
nexus = UnifiedNexus("App", expose="mcp")   # MCP Only (Docs & REST disabled)

2. Tool-Level Exposure

# REST Only
@nexus.api_only(path="/internal-metrics")
def get_metrics():
    return {"cpu": 15}

# MCP Only
@nexus.mcp_only(tool_type=ToolSafetyLevel.READ_ONLY)
def search_agent_memory(query: str):
    return {"matches": []}

# Dual REST + MCP
@nexus.universal_tool(path="/calculate", expose="both")
def calculate(a: int, b: int):
    return {"sum": a + b}

🏢 FastAPI Microservice Mounting

Attach UnifiedNexus directly onto existing production FastAPI applications with existing database pools and custom lifespans:

from contextlib import asynccontextmanager
from fastapi import FastAPI
from unified_nexus import UnifiedNexus

@asynccontextmanager
async def database_lifespan(app: FastAPI):
    print("🔌 Connecting PostgreSQL pool...")
    yield
    print("🛑 Disconnecting PostgreSQL pool...")

existing_app = FastAPI(title="OrderService", lifespan=database_lifespan)
nexus = UnifiedNexus("OrderNexus")

# Mounts routes, normalizes /mcp mount path, and preserves database lifespan:
nexus.attach_to(
    existing_app,
    api_prefix="/api/v1",  # REST routes prefixed: /api/v1/...
    mcp_path="/",          # MCP accessible at /mcp without /mcp/mcp duplication
    mode="both",
)

🧪 Tested Compatibility Matrix

📄 License

Apache License 2.0. See LICENSE for details.