Connection Manager

The Connection Manager handles the lifecycle of downstream MCP server connections, merges their tool schemas into a unified namespace, and supports cross-organization agent delegation.

How It Works

ConnectionManager (`connection_manager.py`)

The ConnectionManager manages stdio connections to downstream MCP servers:

Spawn — Starts a child process for each registered downstream server
Track — Monitors connection health and restarts on failure
Teardown — Gracefully shuts down connections when the workspace is deactivated

Each downstream connection is represented by a DownstreamConnection dataclass with:

Server name and command
Process handle and transport
Health status and last heartbeat

SchemaMerger (`schema_merger.py`)

The SchemaMerger aggregates tool lists from multiple downstream servers and prefixes names to prevent collisions:

Downstream: Salesforce → tools: [update_contact, get_account, ...]
Downstream: Slack → tools: [post_message, create_channel, ...]

Merged: [salesforce.update_contact, salesforce.get_account,
         slack.post_message, slack.create_channel, ...]

Features:

Namespace prefixing — Each tool gets a server_name.tool_name prefix
Description sanitization — Tool descriptions are sanitized to prevent injection (_sanitize_tool_description())
Input schema sanitization — Input schemas are validated and cleaned (_sanitize_input_schema())
TTL cache — Merged schema is cached for 5 minutes to meet NFR-AEG-01 latency targets

Delegation (`delegation.py`)

Cross-organization agent delegation for the agentic web:

Signing — Delegation tokens are cryptographically signed
Rate limiting — Per-delegator rate limits prevent abuse
Inbound handling — Validates incoming delegation tokens from partner organizations
Outbound handling — Signs and sends delegation requests to partner organizations

Delegation supports glob-based tool pattern matching (e.g., delegate access to salesforce.read_* but not salesforce.delete_*).

Architecture

AI Agent
    ↓
Aegis MCP Gate
    ↓
Schema Merger (cached, 5-min TTL)
    ├── Connection: Salesforce MCP
    ├── Connection: Slack MCP
    ├── Connection: Custom API MCP
    └── Delegation: Partner Org Aegis

Configuration

Downstream connections are registered via the REST API:

POST /api/v1/aegis/connections
{
  "workspace_id": "ws-1",
  "name": "salesforce",
  "command": ["python", "-m", "salesforce_mcp_server"],
  "env": {"SALESFORCE_TOKEN": "encrypted:..."}
}

Connection credentials are encrypted at rest using Fernet encryption (crypto.py).

Technical Details

Schema merger sanitization prevents downstream servers from injecting malicious content into tool descriptions
The 5-minute TTL cache balances freshness against latency requirements
Delegation uses fnmatch for tool pattern matching, consistent with policy engine patterns
Connection health is monitored via heartbeats; unhealthy connections trigger automatic restart
The escalation_notifier.py sends notifications when actions require human approval
The identity_adapter.py and external_identity_verifier.py integrate with external IAM/NHI systems for agent identity resolution

How It Works