Architecture & Pipeline

AegisMCP achieves high performance and strict security by using an explicit Execution Pipeline. Unlike simple libraries that directly execute a python function when a request comes in, Aegis isolates the transport layer from the execution logic.

The Execution Flow

When an MCP client (like an LLM or an IDE) sends a request, it flows through four major layers.

sequenceDiagram
    participant C as MCP Client
    participant T as Transport (Stdio/WS)
    participant M as Middleware Pipeline
    participant E as Tool Executor

    C->>T: JSON-RPC Call
    T->>M: Dispatch Request

    rect rgb(20, 20, 40)
        Note over M: 1. AuthMiddleware (Identity)
        Note over M: 2. RateLimitMiddleware
        Note over M: 3. PolicyMiddleware (RBAC)
    end

    M->>E: Execute with AegisContext
    E-->>M: Result
    M-->>T: Format JSON-RPC Response
    T-->>C: Response

1. Transport Layer

Transports like StdioTransport or the ServerSseTransport (via the Starlette adapter) handle the raw IO streams. They decode bytes into JSON-RPC models and route them to the pipeline.

2. Middleware Pipeline

Before a tool is executed, it must pass through the middlewares. Middlewares intercept the request sequentially: - They can Reject the request immediately (e.g., HTTP 429 Too Many Requests via RedisRateLimiter, or 403 Forbidden). - They can Mutate the context (e.g., AuthMiddleware attaches the Identity object).

3. AegisContext

The context is a frozen, explicitly propagated dataclass. It contains: - trace_id and span_id for observability. - The authenticated caller_identity. - A strict deadline for the request.

@dataclass(frozen=True)
class AegisContext:
    request_id: str
    trace_id: str
    span_id: str
    caller_identity: Identity
    permissions: PermissionSet
    deadline: datetime
    metadata: FrozenMapping

4. Tool Executor

Once the middleware allows the request, the ToolExecutor wraps the function in an asyncio.wait_for timeout barrier driven by the AegisContext.deadline.

It gracefully handles both async def and synchronous def functions. Synchronous functions are automatically offloaded to a thread pool via loop.run_in_executor to prevent blocking the asynchronous event loop.