System Architecture

┌─────────────────┐
│   MCP Client    │  (Claude Desktop, Cursor, etc.)
│   (stdio/HTTP)  │
└────────┬────────┘
         │ 1. Connect with API key
         ▼
┌─────────────────────────────────────────────┐
│     Secure MCP Gateway (gateway.py)         │
│     FastMCP Server on 0.0.0.0:8000/mcp/     │
├─────────────────────────────────────────────┤
│  2. Authentication & Authorization          │
│  3. Configuration Loading (cached)          │
└────────┬────────────────────────────────────┘
         │
         ├─────────────┬──────────────────┐
         ▼             ▼                  ▼
   ┌─────────┐  ┌──────────┐      ┌──────────┐
   │Discovery│  │Guardrails│      │Execution │
   │ Service │  │ (Input)  │      │ Service  │
   └────┬────┘  └────┬─────┘      └────┬─────┘
        │            │                  │
        │            └──────┬───────────┘
        ▼                   ▼
   ┌─────────────────────────────────┐
   │   Gateway Client (client.py)    │
   │   stdio_client() to MCP Servers │
   └───────────┬─────────────────────┘
               │
               ▼
   ┌─────────────────────────┐
   │   Actual MCP Servers    │
   │  (GitHub, Filesystem,   │
   │   Custom servers, etc.) │
   └───────────┬─────────────┘
               │
               ▼
   ┌─────────────────────────┐
   │  Response Processing    │
   │  - Output Guardrails    │
   │  - PII De-anonymization │
   │  - Telemetry Logging    │
   └─────────────────────────┘

mcp = FastMCP(
    name="Enkrypt Secure MCP Gateway",
    host="0.0.0.0",
    port=8000,
    streamable_http_path="/mcp/",
    transport="streamable-http" or "stdio"
)

async def forward_tool_call(server_name, tool_name, args=None, gateway_config=None):
    """Forwards tool calls to the appropriate MCP server."""
    # Connects via stdio_client to target server
    async with stdio_client(StdioServerParameters(...)) as (read, write):
        async with ClientSession(read, write) as session:
            # Tool discovery or execution
            return await session.call_tool(tool_name, arguments=args)

class DiscoveryService:
    async def discover_tools(self, ctx, server_name, ...):
        # 1. Check cache first
        # 2. If tools configured, return them
        # 3. Otherwise, call forward_tool_call
        # 4. Cache discovered tools
        # 5. Return tools with metadata

class SecureToolExecutionService:
    async def execute_secure_tools(self, ctx, server_name, tool_calls, logger):
        # 1. Authenticate and get config
        # 2. For each tool call:
        #    a. Run input guardrails
        #    b. Execute tool
        #    c. Run output guardrails
        # 3. Handle PII redaction/de-anonymization

class TimeoutManager:
    def get_timeout(self, operation_type: str) -> float
    def track_operation(self, operation_id, operation_type, start_time)
    def complete_operation(self, operation_id, success)
    def get_metrics() -> Dict

credentials = get_gateway_credentials(ctx)
auth_result = await auth_manager.authenticate(ctx)

gateway_config = await get_local_mcp_config(
    gateway_key, project_id, user_id
)

tools = await discovery_service.discover_tools(ctx, server_name)

results = await secure_tool_execution_service.execute_secure_tools(
    ctx, server_name, tool_calls, logger
)

{
  "command": "mcp",
  "args": ["run", "/path/to/gateway.py"],
  "env": {
    "ENKRYPT_GATEWAY_KEY": "your-api-key",
    "MCP_TRANSPORT": "stdio"
  }
}

# Gateway runs on port 8000
curl -X POST http://gateway:8000/mcp/ \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -d '{"method": "tools/list"}'

apikeys = {
    "YOUR_API_KEY": {
        "project_id": "PROJECT_ID",
        "user_id": "USER_ID",
        "created_at": "2025-01-01T00:00:00"
    }
}

# "GITHUB_TOKEN=abc123def456" -> "GITHUB_TOKEN=abc***456"

def hash_key(key):
    return hashlib.md5(key.encode()).hexdigest()

extra = {
    "custom_id": "...",
    "server_name": "github_server",
    "project_id": "abc-123",
    "user_id": "user-456",
    "tool_name": "search_code"
}
logger.info("Tool executed", extra=extra)

with tracer.start_as_current_span("tool_execution") as span:
    span.set_attribute("server_name", server_name)
    span.set_attribute("tool_name", tool_name)

{
  "enkrypt_mcp_use_external_cache": true,
  "enkrypt_cache_host": "redis.example.com",
  "enkrypt_cache_port": 6379
}