AegisMCP 🛡️

AegisMCP is an opinionated, high-performance Model Context Protocol (MCP) framework designed for enterprise scale.

📚 Read the Full Documentation Here

If other minimal frameworks are like Flask, AegisMCP is like Django. It provides a robust kernel, integrated zero-dependency abstractions, and high-performance asynchronous orchestration to scale your multi-agent architecture.

---

The Enterprise Standard

Other MCP libraries are built for weekend prototypes—they give you a tool in 5 lines of code, but leave you completely exposed in production.

AegisMCP is built for Fortune 500 deployments. We don't bolt security on at the end. Aegis provides a deeply integrated, mathematically safe ExecutionPipeline that natively handles:

• Role-Based Access Control (RBAC) to restrict sensitive tool usage.
• Rate Limiting to prevent LLMs from running up your API bills.
• Deterministic Sagas to automatically roll back multi-step AI tasks when failures occur.
• Multi-Agent Meshes so intelligent agents can serve tools to other agents.

If you are building a weekend toy, use a minimal lightweight framework. If you are deploying an AI Agent into a secure corporate network, use AegisMCP.

---

Core Features

• Parallel Tool Execution: Native asyncio.gather tool orchestrations vastly reduce latency during dense multi-tool generation.
• AegisContext Everywhere: Explicit context propagation eliminates race conditions and ties every tool call to standard IDs (request_id, trace_id, span_id).
• Zero Mandatory Dependencies: The core framework requires only pydantic and anyio. Run a basic stdio server in a 50MB container.
• First-Class Security: Customizable Auth, RBAC policy gating, Rate Limiting, and Audit Logging right out of the box.
• Built-In Agent Runtime: Scale your AI via hierarchical sub-agents, deterministic Saga workflows, and multi-tool selection logic via the built-in Layer 5/6 engines.

Installation

pip install aegismcp-core

Or install with all optional extensions (HTTP transports, observability SDKs, vector database adapters):

pip install "aegismcp-core[all]"

Quick Start (Stdio & HTTP)

The simplest AegisMCP application runs entirely over Stdio.

import asyncio
from aegismcp.server.app import AegisMCP

app = AegisMCP("HelloWorld")

@app.tool(description="Say hello")
async def say_hello(name: str) -> str:
    return f"Hello, {name}!"

if __name__ == "__main__":
    asyncio.run(app.run_stdio())

AegisMCP is not limited to Stdio. You can instantly expose it via HTTP Server-Sent Events (SSE) by wrapping it in an ASGI application using our robust Starlette Adapter:

import uvicorn
from aegismcp.server.app import AegisMCP
from aegismcp.adapters.starlette.transport import create_starlette_app

app = AegisMCP("HttpServer")
asgi_app = create_starlette_app(app)

if __name__ == "__main__":
    uvicorn.run(asgi_app, host="0.0.0.0", port=8000)

Local Unit Testing

Unlike other frameworks, AegisMCP enforces strict context propagation. When the ExecutionPipeline runs your tools over a network, it automatically injects a hardened AegisContext.

If you want to bypass the server and directly test your tools locally in Python, you simply construct the context yourself:

import asyncio
from datetime import datetime, timedelta, UTC
from aegismcp.server.app import AegisMCP
from aegismcp.kernel.context import AegisContext, Identity

app = AegisMCP("TestServer")

@app.tool()
async def say_hello(ctx: AegisContext, name: str) -> str:
    return f"Hello, {name}! AegisMCP is working perfectly!"

async def main():
    # Manually construct the security context for local testing
    ctx = AegisContext(
        request_id="test-req-001",
        trace_id="trace-123",
        span_id="span-123",
        caller_identity=Identity(id="TestUser", type="user"),
        permissions=frozenset(),
        deadline=datetime.now(UTC) + timedelta(minutes=5),
        metadata={},
        baggage={}
    )
    
    result = await say_hello(ctx, name="Ujjwal")
    print(result)

if __name__ == "__main__":
    asyncio.run(main())

Enterprise Security (API Keys + RBAC)

import asyncio
from aegismcp.server.app import AegisMCP
from aegismcp.security.auth.apikey import ApiKeyAuth
from aegismcp.security.policy.rbac import RBACPolicyEngine

# Define Identity verification logic
async def verify_key(key: str):
    if key == "secret": return {"id": "user123", "roles": ["admin"]}
    return None

auth = ApiKeyAuth(verify_key)
policy = RBACPolicyEngine(role_permissions={"admin": frozenset(["users:read", "users:write"])})

app = AegisMCP("SecureApp", auth=auth, policy=policy)

@app.tool(
    description="Secure data retrieval",
    required_permissions=frozenset(["users:read"])
)
async def get_secure_data() -> str:
    return "Sensitive Information"

Deterministic Workflows

import asyncio
from aegismcp.server.app import AegisMCP
from aegismcp.kernel.context import AegisContext

app = AegisMCP("WorkflowApp")

class Step1:
    async def execute(self, ctx: AegisContext): return "Step 1"
    async def compensate(self, ctx: AegisContext): print("Rolling back Step 1")

@app.workflow("onboarding")
async def onboarding_saga(ctx: AegisContext):
    # This automatically rolls back on failure in reverse order
    return await app.workflow_engine.execute_saga([
        (Step1(), (), {}),
        # ... Add Step 2 that throws Error
    ], ctx)

Advanced Agent Runtimes

Multi-Agent Mesh & Runtime

AegisMCP extends the Model Context Protocol far beyond simple remote procedure calls.

AegisAgent

The AegisAgent class acts as the brains of your system. You provide it a ModelProvider (like Anthropic or OpenAI) and a list of tool names. It autonomously iterates through an execution loop, deciding which tools to call.

The agent_as_tool Pattern

In a true Enterprise Multi-Agent Mesh, agents need to talk to other agents. AegisMCP allows you to instantly convert an entire AegisAgent into a standard MCP tool using the agent_as_tool adapter.

This allows a top-level Router Agent to call a "Database Analyst Agent" simply by executing a tool call, seamlessly passing contexts down the tree.

Security Model

AegisMCP treats security as a first-class citizen using the ExecutionPipeline and Middleware architecture.

Role-Based Access Control (RBAC)

When defining a tool, you can specify required_permissions:

@app.tool(required_permissions={"admin:write"})
async def delete_user(user_id: str):
    pass

The pipeline verifies the identity attached to the AegisContext.

API Key Authentication

AegisMCP provides built-in ApiKeyMiddleware. It extracts tokens from headers (in HTTP transports) or connection payloads (in WebSockets).

Distributed Rate Limiting

The RateLimitMiddleware ensures that a single identity cannot spam the RPC server, preventing DOS attacks. For production enterprise deployments, AegisMCP includes a distributed RedisRateLimiter powered by an atomic token-bucket Lua script.

Workflows & Deterministic Sagas

AegisMCP uses the WorkflowEngine to implement the Saga Pattern.

Why Sagas?

In a multi-agent or multi-tool system, tasks often require multiple sequential steps (e.g., Book Flight, Reserve Hotel, Charge Credit Card).

If "Charge Credit Card" fails, you can't simply throw an error—you must roll back the hotel and flight!

Implementation

class BookFlightStep:
    async def execute(self, ctx: AegisContext):
        return await flight_api.book()
        
    async def compensate(self, ctx: AegisContext):
        # Triggered automatically if ANY subsequent step fails!
        await flight_api.cancel()

AegisMCP executes these sagas deterministically, ensuring your enterprise system always returns to a stable state.

Author

Ujjwal Jagtap

• Email: ujjwaljagtap7@gmail.com
• GitHub: Ujje421

Built with ❤️ for the AI integration community.