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 — Records connection health (does not restart on failure). POST /api/v1/praxom/connections/{id}/health probes an HTTP/SSE URL with MCP initialize + tools/list and records healthy, unreachable, or error. stdio is error — the REST route does not spawn the stored command.
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 30 seconds (SchemaMerger constructor default) 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 — A governor sends via POST /api/v1/praxom/delegation/outbound; the request is signed and attested (intent before send, result after)

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

Architecture

AI Agent
    ↓
Praxom MCP Gate
    ↓
Schema Merger (cached, 30 s TTL)
    ├── Connection: Salesforce MCP
    ├── Connection: Slack MCP
    ├── Connection: Custom API MCP
    └── Delegation: Partner Org Praxom

Configuration

Downstream connections are registered via the REST API:

POST /api/v1/praxom/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 30-second TTL cache balances freshness against latency requirements
Delegation uses fnmatch for tool pattern matching, consistent with policy engine patterns
Connection health is recorded; heartbeats do not trigger a restart. The health route probes with MCP initialize + tools/list as above.
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 WorksCopy Anchor Link