API Reference

This section contains the auto-generated API reference for the AegisMCP framework.

AegisMCP Application

aegismcp.server.app

AegisMCP

Source code in aegismcp\server\app.py

class AegisMCP:
    def __init__(
        self,
        name: str,
        version: str = "1.0.0",
        description: str = "",
        auth: Any = None,
        policy: Any = None,
        audit: Any = None,
        rate_limit: int | None = None,
        telemetry_metrics: Any = None,
        telemetry_tracer: Any = None,
    ):
        self.name = name
        self.version = version
        self.description = description
        self.tools: dict[str, ToolDescriptor] = {}

        from aegismcp.workflow.engine import WorkflowEngine

        self.workflow_engine = WorkflowEngine()

        from aegismcp.execution.pipeline import Middleware

        self.middlewares: list[Middleware] = []
        if telemetry_metrics and telemetry_tracer:
            from aegismcp.execution.middleware.observability import ObservabilityMiddleware

            self.middlewares.append(ObservabilityMiddleware(telemetry_metrics, telemetry_tracer))
        if rate_limit is not None:
            from aegismcp.execution.middleware.ratelimit import RateLimitMiddleware

            self.middlewares.append(RateLimitMiddleware(rate_limit))
        if auth:
            from aegismcp.execution.middleware.auth import AuthenticationMiddleware

            self.middlewares.append(AuthenticationMiddleware(auth))
        if policy:
            from aegismcp.execution.middleware.auth import AuthorizationMiddleware

            self.middlewares.append(AuthorizationMiddleware(policy))
        if audit:
            from aegismcp.execution.middleware.audit import AuditMiddleware

            self.middlewares.append(AuditMiddleware(audit))

        self.executor = ToolExecutor()
        self.pipeline = ExecutionPipeline(self.middlewares, self.executor)

    def tool(self, **kwargs: Any) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
        """Register a tool. Delegates to the @tool decorator."""

        def decorator(fn: Callable[..., Any]) -> Callable[..., Any]:
            decorated_fn: Callable[..., Any] = tool(**kwargs)(fn)
            descriptor = getattr(decorated_fn, "__aegis_tool__")
            self.tools[descriptor.name] = descriptor
            return decorated_fn

        return decorator

    def workflow(self, name: str | None = None) -> Any:
        def decorator(fn: Any) -> Any:
            wf_name = name or fn.__name__
            self.workflow_engine.register(wf_name, fn)
            return fn

        return decorator

    async def run_stdio(self) -> None:
        """Run the server using stdio transport."""
        codec = ProtocolCodec()
        transport = StdioTransport(codec)
        await self.run_transport(transport)

    async def run_transport(self, transport: Transport) -> None:
        from aegismcp.protocol.messages import JSONRPCResponse, JSONRPCRequest

        await transport.start()
        try:
            async for message in transport.receive():
                if isinstance(message, JSONRPCRequest):
                    # Basic mock implementation to unblock transport tests
                    # A real router would parse the method, find the tool, run the pipeline, etc.
                    response = JSONRPCResponse(
                        jsonrpc="2.0",
                        id=message.id,
                        result={"success": True}
                    )
                    await transport.send(response)
        finally:
            await transport.stop()

run_stdio() -> None async

Run the server using stdio transport.

Source code in aegismcp\server\app.py

async def run_stdio(self) -> None:
    """Run the server using stdio transport."""
    codec = ProtocolCodec()
    transport = StdioTransport(codec)
    await self.run_transport(transport)

tool(**kwargs: Any) -> Callable[[Callable[..., Any]], Callable[..., Any]]

Register a tool. Delegates to the @tool decorator.

Source code in aegismcp\server\app.py

def tool(self, **kwargs: Any) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
    """Register a tool. Delegates to the @tool decorator."""

    def decorator(fn: Callable[..., Any]) -> Callable[..., Any]:
        decorated_fn: Callable[..., Any] = tool(**kwargs)(fn)
        descriptor = getattr(decorated_fn, "__aegis_tool__")
        self.tools[descriptor.name] = descriptor
        return decorated_fn

    return decorator

Security Context

aegismcp.kernel.context

AegisContext dataclass

The frozen, explicitly-propagated value object that carries request state.

This solves tracing, security, testing, and debugging simultaneously.

Source code in aegismcp\kernel\context.py

@dataclass(frozen=True)
class AegisContext:
    """
    The frozen, explicitly-propagated value object that carries request state.

    This solves tracing, security, testing, and debugging simultaneously.
    """

    request_id: str
    trace_id: str
    span_id: str
    caller_identity: Identity
    permissions: PermissionSet
    deadline: datetime
    metadata: FrozenMapping
    baggage: FrozenMapping

    def with_deadline(self, new_deadline: datetime) -> "AegisContext":
        """Return a new context with a tighter deadline (cannot extend)."""
        if new_deadline > self.deadline:
            # Can only tighten deadlines, not extend them
            return self
        return replace(self, deadline=new_deadline)

    def with_span(self, new_span_id: str) -> "AegisContext":
        """Return a new context with a new span ID for tracing sub-operations."""
        return replace(self, span_id=new_span_id)

    def with_metadata(self, new_metadata: Mapping[str, Any]) -> "AegisContext":
        """Return a new context with merged metadata."""
        merged = dict(self.metadata)
        merged.update(new_metadata)
        return replace(self, metadata=merged)

with_deadline(new_deadline: datetime) -> AegisContext

Return a new context with a tighter deadline (cannot extend).

Source code in aegismcp\kernel\context.py

def with_deadline(self, new_deadline: datetime) -> "AegisContext":
    """Return a new context with a tighter deadline (cannot extend)."""
    if new_deadline > self.deadline:
        # Can only tighten deadlines, not extend them
        return self
    return replace(self, deadline=new_deadline)

with_metadata(new_metadata: Mapping[str, Any]) -> AegisContext

Return a new context with merged metadata.

Source code in aegismcp\kernel\context.py

def with_metadata(self, new_metadata: Mapping[str, Any]) -> "AegisContext":
    """Return a new context with merged metadata."""
    merged = dict(self.metadata)
    merged.update(new_metadata)
    return replace(self, metadata=merged)

with_span(new_span_id: str) -> AegisContext

Return a new context with a new span ID for tracing sub-operations.

Source code in aegismcp\kernel\context.py

def with_span(self, new_span_id: str) -> "AegisContext":
    """Return a new context with a new span ID for tracing sub-operations."""
    return replace(self, span_id=new_span_id)

Identity dataclass

Represents the authenticated caller identity.

Source code in aegismcp\kernel\context.py

@dataclass(frozen=True)
class Identity:
    """Represents the authenticated caller identity."""

    id: str
    type: str  # e.g., "user", "service", "anonymous"
    attributes: FrozenMapping = field(default_factory=dict)

create_anonymous_context(request_id: str, trace_id: str, span_id: str, deadline: datetime) -> AegisContext

Create a context for an unauthenticated request.

Source code in aegismcp\kernel\context.py

def create_anonymous_context(
    request_id: str, trace_id: str, span_id: str, deadline: datetime
) -> AegisContext:
    """Create a context for an unauthenticated request."""
    return AegisContext(
        request_id=request_id,
        trace_id=trace_id,
        span_id=span_id,
        caller_identity=Identity(id="anonymous", type="anonymous", attributes={}),
        permissions=frozenset(),
        deadline=deadline,
        metadata={},
        baggage={},
    )

Tools

aegismcp.tools.descriptor

aegismcp.tools.decorator

tool(name: str | None = None, description: str | None = None, timeout: float = 30.0, max_retries: int = 0, retry_delay: float = 1.0, is_idempotent: bool = False, permissions: frozenset[str] = frozenset(), audit: str = 'METADATA', cache_ttl: int | None = None, tags: list[str] | None = None) -> Callable[[Callable[..., Any]], Callable[..., Any]]

Decorator to register a function as an MCP tool.

Source code in aegismcp\tools\decorator.py

def tool(
    name: str | None = None,
    description: str | None = None,
    timeout: float = 30.0,
    max_retries: int = 0,
    retry_delay: float = 1.0,
    is_idempotent: bool = False,
    permissions: frozenset[str] = frozenset(),
    audit: str = "METADATA",
    cache_ttl: int | None = None,
    tags: list[str] | None = None,
) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
    """Decorator to register a function as an MCP tool."""

    def decorator(fn: Callable[..., Any]) -> Callable[..., Any]:
        fn_name = name or fn.__name__
        fn_desc = description or fn.__doc__ or ""

        input_schema = generate_json_schema(fn)

        descriptor = ToolDescriptor(
            name=fn_name,
            description=fn_desc,
            input_schema=input_schema,
            output_schema=None,  # Simplified for now
            timeout_seconds=timeout,
            max_retries=max_retries,
            retry_delay_seconds=retry_delay,
            is_idempotent=is_idempotent,
            required_permissions=permissions,
            audit_level=audit,
            fn=fn,
        )

        # Attach descriptor to the function so it can be picked up by the app registry
        setattr(fn, "__aegis_tool__", descriptor)
        return fn

    return decorator

Execution & Workflows

aegismcp.execution.pipeline

aegismcp.execution.executor

aegismcp.workflow.engine

Transports

aegismcp.transports.stdio