Intake Adapters

Intake adapters normalize agent actions from different protocols into the common format that the Aegis Policy Decision Point evaluates. This allows Aegis to validate actions regardless of how the agent communicates with downstream systems.

How It Works

The MCP Gate (mcp_gate.py) is the primary intake adapter, handling the Model Context Protocol natively. Additional intake surfaces are planned for OpenAPI and FHIR SMART protocols.

MCP Intake (Primary)

The AegisMCPGate class is the enforcement point for all MCP tool calls:

1. Identity resolution — Extracts agent identity from headers (extract_agent_identity())
2. Tool classification — Determines if the call is a read (pass-through) or action (evaluate)
3. Policy evaluation — Calls the policy engine for ALLOW/DENY/REWRITE/ESCALATE verdict
4. Rewrite application — For REWRITE verdicts, applies safe transformations to the payload
5. Attestation recording — Records the decision with Ed25519 signature
6. Error formatting — Returns tiered error responses (agents get sanitized errors)

gate = get_mcp_gate()
result = await gate.evaluate(
    tool_name="salesforce.update_contact",
    arguments={"contact_id": "123", "email": "new@example.com"},
    headers=request.headers
)
# result.verdict: "allow" | "deny" | "rewrite" | "escalate"
# result.arguments: potentially modified arguments
# result.attestation_id: UUID of the recorded attestation

OpenAPI Intake (Planned)

Will normalize REST API calls into the Aegis action format, allowing policy evaluation for agents that interact via HTTP APIs rather than MCP.

FHIR SMART Intake (Planned)

Will normalize FHIR SMART on FHIR actions for healthcare use cases, mapping FHIR operations (read, create, update, delete) to Aegis action patterns.

Architecture

AI Agent
    ├── MCP Protocol → AegisMCPGate (active)
    ├── REST API → OpenAPI Intake (planned)
    └── FHIR SMART → FHIR Intake (planned)
    ↓
Normalized Action
    ↓
Aegis PDP (policy evaluation)
    ↓
Verdict + Attestation

The Enforcement Gate

The GateResult dataclass carries the outcome:

@dataclass
class GateResult:
    verdict: str           # "allow", "deny", "rewrite", "escalate"
    arguments: dict        # Original or modified arguments
    attestation_id: str    # UUID of the recorded attestation
    reasons: list          # Constraint evaluation details
    impact_set: dict       # Blast radius computation

Error formatting (_format_error()) strips internal details before returning to agents. The agent sees a structured error with the verdict and reason, but not internal constraint IDs, database query details, or stack traces.

Configuration

The MCP Gate is automatically initialized when the Aegis API starts. No additional configuration is needed beyond the standard Aegis setup.

Identity resolution can be customized via the identity_adapter.py module to integrate with external IAM/NHI systems (Azure AD, Okta, custom identity providers).

Technical Details

• The MCP Gate is a singleton retrieved via get_mcp_gate() for consistent state
• Agent identity extraction supports multiple header formats for compatibility with different MCP clients
• Error sanitization uses sanitize_error_internal() to prevent information leakage
• The gate handles async evaluation — all I/O operations (database queries, downstream calls) are non-blocking
• Escalation notifications are sent via escalation_notifier.py when a verdict requires human review