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.

Distributed Rate Limiting

If you are deploying AegisMCP across multiple workers or containers, the default in-memory rate limiter will not suffice. Instead, use the RedisRateLimiter to maintain a distributed token bucket.

import redis.asyncio as redis
from aegismcp.server.app import AegisMCP
from aegismcp.adapters.redis.ratelimit import RedisRateLimiter

# Connect to your Redis cluster
redis_client = redis.Redis.from_url("redis://localhost")

# 60 requests per minute per user identity
rate_limiter = RedisRateLimiter(redis_client, tokens_per_minute=60)

app = AegisMCP("EnterpriseApp", rate_limit=rate_limiter)

Custom Middleware Deep Dive

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

Middleware functions use a classic __call__ chaining pattern.

from typing import Any
from aegismcp.kernel.context import AegisContext
from aegismcp.execution.pipeline import Middleware, ToolHandler
from aegismcp.tools.descriptor import ToolDescriptor

class CustomAuditMiddleware(Middleware):
    """
    A custom middleware that logs the start and end of every tool call.
    """
    async def __call__(
        self, 
        inputs: Any, 
        ctx: AegisContext, 
        descriptor: ToolDescriptor,
        next_handler: ToolHandler
    ) -> Any:

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

        try:
            # 2. CALL next middleware (or the Tool Executor)
            result = await next_handler(inputs, ctx, descriptor)

            # 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())