Enterprise Security

Security in AegisMCP is not an afterthought. It is woven into the very fabric of the ExecutionPipeline.

The Pipeline Architecture

When a request arrives at the AegisMCP server, it does not immediately execute a tool. It must first pass through the Middleware Pipeline:

1. Authentication Middleware: Validates API keys, JWTs, or other credentials and resolves them into an Identity.
2. Rate Limit Middleware: Checks the Identity against a quota to prevent DOS attacks and exorbitant LLM costs.
3. Policy Middleware (RBAC): Checks the Identity's roles against the required_permissions of the target Tool.

Role-Based Access Control (RBAC)

You can protect any tool by defining required_permissions.

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

# 1. Define how to verify API keys and map them to Roles
async def verify_key(key: str):
    if key == "sk-admin-secret": 
        return {"id": "user123", "roles": ["admin"]}
    return None

auth = ApiKeyAuth(verify_key)

# 2. Define what permissions each Role grants
policy = RBACPolicyEngine(role_permissions={
    "admin": frozenset(["users:read", "users:write", "system:reboot"])
})

# 3. Mount Security to the App
app = AegisMCP("SecureApp", auth=auth, policy=policy)

# 4. Protect Tools
@app.tool(
    description="Deletes a user account",
    required_permissions=frozenset(["users:write"])
)
async def delete_user(user_id: str) -> str:
    return f"User {user_id} deleted securely."

If an LLM without the admin role attempts to call delete_user, the ExecutionPipeline instantly rejects the request with a 403 Forbidden error before the function code is ever reached.

Custom Middleware Deep Dive

AegisMCP allows you to completely customize the pipeline by writing your own Middleware.

Middleware functions use the Onion Architecture pattern via yield.

from typing import Any
from collections.abc import AsyncGenerator
from aegismcp.kernel.context import AegisContext
from aegismcp.execution.middleware.base import Middleware
from aegismcp.tools.descriptor import ToolDescriptor

class CustomAuditMiddleware(Middleware):
    """
    A custom middleware that logs the start and end of every tool call,
    including how long it took.
    """
    async def process(
        self, 
        inputs: Any, 
        ctx: AegisContext, 
        descriptor: ToolDescriptor
    ) -> AsyncGenerator[Any, Any]:

        # 1. PRE-EXECUTION (Before the tool runs)
        print(f"[AUDIT] Starting {descriptor.name} for user {ctx.caller_identity.id}")

        try:
            # 2. YIELD to the next middleware (or the Tool Executor)
            result = yield inputs

            # 3. POST-EXECUTION (After the tool finishes successfully)
            print(f"[AUDIT] Successfully completed {descriptor.name}")
            return result

        except Exception as e:
            # 4. ERROR HANDLING (If the tool crashed)
            print(f"[AUDIT] Tool {descriptor.name} failed with error: {e}")
            raise

To use it, just add it to your app:

app = AegisMCP("SecureApp", audit=CustomAuditMiddleware())