Unified Nexus

The Problem

Stop writing the same logic twice.

Traditional AI backends force developers to write a FastAPI route, then manually re-wrap it as an MCP tool. Over time, schemas drift, authentication gets bypassed, and internal DB handles leak into LLM prompts.

❌ Vulnerable & Drifting Stack

# Two definitions. Different validations.
routes/orders.py     ← HTTP with Depends(get_user)
mcp/orders.py        ← Copied MCP tool (bypasses auth)
models/orders.py     ← Schema drift between OpenAPI & LLM

# Critical Vulnerabilities:
• LLMs prompted to supply user_id / credentials
• Silent security bypasses in agent layer
• Synchronous tools starve the async event loop

✅ UnifiedNexus v0.2.0

# One definition. Fail-closed security boundaries.
main.py              ← One function. One decorator.
@nexus.universal_tool← Dual interfaces automatically.

# Verified Invariants:
✔ FastAPI Depends() rejected from MCP schemas
✔ Sensitive credentials masked with ***
✔ Threadpool offloading for slow sync tools
✔ Single lifespan & normalized /mcp mount

Features

Enterprise AI tool serving.
Zero bloat.

⚡

Single Source of Truth

Define business logic once. REST and MCP share identical Pydantic validation, schemas, and docstrings — forever synchronized.

🛡️

Fail-Closed Security

Strict boundary separation. FastAPI Depends() and credential parameters are rejected from MCP tool arguments at startup.

🧩

Bidirectional Plugins

NexusPlugin intercepts both REST and MCP requests using unified NexusContext for rate limiting, auditing, and tracing.

🤖

AI Safety Invariants

Classify tools with ToolSafetyLevel (READ_ONLY, ACTION, DESTRUCTIVE), human approvals, and payload bounding.

🏢

Microservice Mount

Mount onto existing FastAPI apps via nexus.attach_to(). Normalized /mcp path and clean lifespan composition.

🤝

LangServe Collaboration

Deploy LangChain LCEL runnables alongside FastMCP tools on a single ASGI application without conflicts.

Interactive Code Walkthrough

Up and running in seconds.

Explore how to install, define universal tools, register plugins, and mount to microservices.

# Install via pip
pip install unified-nexus

# Verify installation
python -c "import unified_nexus; print(unified_nexus.__name__)"

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

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

class UserQuery(BaseModel):
    user_id: int = Field(..., description="The customer account ID")

@nexus.universal_tool(path="/customer", tool_type=ToolSafetyLevel.READ_ONLY, idempotent=True)
def get_customer(query: UserQuery):
    """
    Returns user account status and subscription tier.
    Auto-published to OpenAPI Swagger UI and Claude / MCP inspector tools.
    """
    return {"id": query.user_id, "status": "Active", "tier": "Enterprise"}

app = nexus.finalize()

# Run server with uvicorn
uvicorn main:app --host 127.0.0.1 --port 8000

# 🌐 REST API & OpenAPI Swagger Docs:
http://localhost:8000/docs

# 🤖 MCP SSE Transport Endpoint:
http://localhost:8000/mcp

# 🔍 Inspect MCP Tools interactively:
npx -y @modelcontextprotocol/inspector http://localhost:8000/mcp

from unified_nexus import NexusPlugin, NexusContext

class AuditTrailPlugin(NexusPlugin):
    async def before_execute(self, ctx: NexusContext, kwargs: dict):
        print(f"[{ctx.request_id}] {ctx.transport.upper()} calling {ctx.tool_name}")

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

nexus.add_plugin(AuditTrailPlugin())

# Destructive tool requiring human approval and bounded output
@nexus.universal_tool(
    path="/terminate-server",
    tool_type=ToolSafetyLevel.DESTRUCTIVE,
    requires_approval=True,
    max_output_chars=5000,
    timeout=10.0,
)
def terminate_server(server_id: str, confirmation: str):
    """Terminates a production server. Requires human approval."""
    return {"server": server_id, "status": "terminated"}

from fastapi import FastAPI
from unified_nexus import UnifiedNexus

existing_app = FastAPI(title="OrderMicroservice")
nexus = UnifiedNexus("OrderNexus")

# Mounts REST at /api/v1 and MCP at /mcp cleanly preserving lifespans:
nexus.attach_to(existing_app, api_prefix="/api/v1", mcp_path="/", mode="both")

Enterprise Boundaries

Dual Transport Isolation

REST authentication does not secure MCP tools, and MCP authentication does not protect REST routes. UnifiedNexus enforces fail-closed isolation.

Security Boundary	REST Behavior	MCP (AI Agent) Behavior
FastAPI `Depends()`	Executed via HTTP Request headers	Rejected with `SecurityConfigurationError`
Credential Parameters	Extracted via Authorization headers	Blocked from LLM tool arguments (no spoofing)
Internal Exceptions	Formatted as HTTP JSON response	Sanitized (passwords, tokens, DB URLs masked)
Router Isolation	Standard `include_router`	Protected routers excluded unless authorized

AI Safety

Operational Invariants

Prevent agent swarms, prompt injection floods, and unintended destructive side effects.

Invariant	Configuration	Purpose & Protection
Tool Safety Levels	`tool_type=ToolSafetyLevel.READ_ONLY \| ACTION \| DESTRUCTIVE`	Explicit capability tiering for agent planners
Human Approval Hook	`requires_approval=True`, `approval_hook=...`	Blocks high-risk actions without supervisor signature
Idempotency	`idempotent=True`	Signals safe automated retries for AI agents
Bounded Payloads	`max_input_chars`, `max_output_chars`	Guards context windows against prompt floods
Threadpool Offload	Automatic (Starlette `run_in_threadpool`)	Prevents slow sync tools from starving event loop

AI Integrations

Connect Any AI Host in Seconds

Your FastMCP endpoint operates seamlessly with every major AI ecosystem out of the box.

🟣 Claude Desktop

{
  "mcpServers": {
    "my-api": {
      "url": "http://localhost:8000/mcp",
      "transport": "streamable-http"
    }
  }
}

🔵 Cursor / VS Code

# In Cursor MCP Settings:

Server URL:
http://localhost:8000/mcp

Transport:
streamable-http

🔍 MCP Inspector

# Test & inspect all tools:

npx -y @modelcontextprotocol/inspector \
  http://localhost:8000/mcp