Skip to content

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 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
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 transport.start()

        try:
            async for message in transport.receive():
                # To be implemented with proper execution pipeline
                pass
        finally:
            await transport.stop()

run_stdio() -> None async

Run the server using stdio transport.

Source code in aegismcp\server\app.py 
80
81
82
83
84
85
86
87
88
89
90
91
async def run_stdio(self) -> None:
    """Run the server using stdio transport."""
    codec = ProtocolCodec()
    transport = StdioTransport(codec)
    await transport.start()

    try:
        async for message in transport.receive():
            # To be implemented with proper execution pipeline
            pass
    finally:
        await transport.stop()

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

Register a tool. Delegates to the @tool decorator.

Source code in aegismcp\server\app.py 
61
62
63
64
65
66
67
68
69
70
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 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
@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 
37
38
39
40
41
42
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 
48
49
50
51
52
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 
44
45
46
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 
11
12
13
14
15
16
17
@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 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
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 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
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