Enterprise MCP Gateway

A modular reverse proxy for Model Context Protocol (MCP) servers. The gateway spawns downstream stdio MCP server subprocesses, aggregates their tools under namespaced names, and exposes a unified MCP server interface to AI clients.

Features

• Multi-server aggregation — combine many MCP servers behind a single endpoint

• Automatic namespacing — tool collisions resolved via server__tool prefixing

• YAML + Pydantic config — validated configuration that fails fast on errors

• Crash detection & auto-restart — monitor subprocess health, restart on crash

• Clean shutdown — SIGINT/SIGTERM handled gracefully, no orphaned processes

• Configurable timeouts — per-server request timeouts with per-request overrides

• Structured JSON logging — all requests and responses logged for observability

• Pure stdio transport — no HTTP dependency, works with any MCP host

• Educational inspection — optional stderr protocol traces for learning and debugging

Related MCP server: MCP Proxy Server

Prerequisites

• Python 3.12 or later

• The downstream MCP servers you want to proxy (e.g. npx, uvx, or any stdio MCP server)

Quickstart

# 1. Clone and install
git clone https://github.com/atifksiddiqui/mcp-gw.git && cd mcp-gw
python3 -m venv .venv && source .venv/bin/activate

# Development (includes pytest, pytest-asyncio, ruff):
pip install -e ".[dev]"

# Production (runtime dependencies only, no editable mode):
pip install .

# 2. Verify it works with the built-in mock servers
python scripts/quick_e2e.py

# 3. Run the gateway (connects an AI client via stdin/stdout)
mcp-gateway gateway_config.yaml

Expected output from step 2:

[OK] Started server 'mock' (pid=12345)
[OK] Initialized: {'name': 'mock-server', 'version': '0.1.0'}
[OK] tools/list returned 2 tool(s):
       - echo: Echo back the provided text.
       - add: Add two integers and return the sum.
[OK] tools/call echo: 'hello world'
[OK] tools/call add(7, 35): 42
[ALL CHECKS PASSED]

Additional verification scripts:

# Multi-server namespace routing
python scripts/verify_multi_server.py

# Example MCP client connecting to the gateway as a subprocess
python scripts/example_client.py gateway_config.yaml

# Educational traffic initiation (offline/online)
python scripts/learn_traffic_init.py

The gateway exits cleanly on EOF, SIGINT, or SIGTERM. See the Configuration section below for customizing gateway_config.yaml.

Architecture

 AI Client (e.g. Claude)
    │  stdin/stdout (JSON-RPC 2.0)
    ▼
┌──────────────────────────────────────────────────┐
│                   MCP Gateway                      │
│                                                    │
│  Config Loader ── Pydantic + YAML validation       │
│  ToolRegistry  ── namespaced tool aggregation     │
│  Routing       ── resolve("srv_a__echo")          │
│                    → ("srv_a", "echo")             │
│  Transports    ── StdioTransport per server        │
│  Monitoring    ── crash detection + auto-restart   │
│  Server Loop   ── MCP server on own stdin/stdout   │
│  Signal Handle ── SIGINT/SIGTERM → clean shutdown  │
└──────┬──────────────────────┬──────────────────────┘
       │ stdin/stdout         │ stdin/stdout
       ▼                      ▼
   ┌───────────┐          ┌───────────┐
   │  Server A │          │  Server B │
   │           │          │           │
   │ echo      │          │ greet     │
   │ add       │          │ multiply  │
   └───────────┘          └───────────┘

The gateway is itself an MCP server. AI clients connect to the gateway's stdin/stdout. The gateway forwards requests to the correct downstream subprocess based on the namespace prefix in tool names (server__tool).

Configuration

Edit gateway_config.yaml to define your downstream servers:

servers:
  - name: mock_a
    command: python3
    args:
      - examples/mock_server.py

  - name: mock_b
    command: python3
    args:
      - examples/mock_server.py
    env:
      MOCK_TOOLS: "alt"
      MOCK_SERVER_NAME: "mock-server-b"

logging:
  level: INFO
  format: json

This config creates two downstream servers. Tools are namespaced as:

• mock_a__echo, mock_a__add (from server A)

• mock_b__greet, mock_b__multiply (from server B)

If both servers define a tool with the same local name (e.g. echo), they remain distinct: mock_a__echo and mock_b__echo.

Each server entry supports:

Real-world example with resiliency enabled:

servers:
  # Filesystem access
  - name: fs
    command: npx
    args: ["-y", "@modelcontextprotocol/server-filesystem", "/home/user/docs"]
    auto_restart: true
    max_restarts: 3

  # GitHub integration
  - name: github
    command: npx
    args: ["-y", "@modelcontextprotocol/server-github"]
    env:
      GITHUB_PERSONAL_ACCESS_TOKEN: "ghp_..."
    timeout: 60

  # SQLite database
  - name: sqlite
    command: uvx
    args: ["mcp-server-sqlite", "--db-path", "/data/app.db"]
    auto_restart: true

With this config, an AI client sees: fs__read_file, fs__write_file, github__create_issue, sqlite__query, etc. — all through one connection, no tool name collisions.

Usage with MCP Clients

The gateway is a standard MCP stdio server. Any MCP-compatible client can connect to it.

Claude Desktop

Add the gateway to your claude_desktop_config.json:

{
  "mcpServers": {
    "gateway": {
      "command": "/path/to/mcp-gw/.venv/bin/mcp-gateway",
      "args": ["/path/to/mcp-gw/gateway_config.yaml"]
    }
  }
}

Cursor / VS Code (MCP extension)

Point your MCP client at the gateway command:

mcp-gateway /path/to/gateway_config.yaml

Programmatic (Python)

import asyncio
from mcp_gateway.gateway.server import Gateway

async def main():
    async with Gateway("gateway_config.yaml") as gw:
        tools = await gw.list_tools()
        print(f"{len(tools)} tools available")
        for t in tools:
            print(f"  {t['name']}: {t.get('description', '')}")

        result = await gw.call_tool("mock_a__echo", {"text": "hello"})
        print(result["content"][0]["text"])

asyncio.run(main())

How Routing Works

When a client calls tools/call with a namespaced name like mock_a__echo:

1. The gateway looks up mock_a__echo in the registry

2. The registry resolves it to ("mock_a", "echo") — the server name and original tool name

3. The gateway finds the StdioTransport for mock_a

4. It forwards tools/call with the original name echo (not the namespaced one)

5. The result is passed back to the client

If the tool doesn't exist in the registry, the gateway returns a JSON-RPC error. If the owning server is down, the gateway returns an internal error.

Resiliency

Process Monitoring

Each downstream subprocess is watched by a background monitor task. When a subprocess exits unexpectedly (not via a normal stop() call):

1. The monitor sets a crashed flag with the exit code

2. Any in-flight requests are rejected with a StdioTransportError

3. The gateway's crash callback is invoked

Auto-Restart

If auto_restart: true is set, the crash callback:

1. Increments the restart counter for that server

2. Calls transport.restart() (stop + start + initialize)

3. Re-discovers tools and re-registers them in the registry

4. The gateway is ready to route calls again

If the restart count exceeds max_restarts, or if auto_restart is false, the server's tools are removed from the registry and the gateway returns errors for calls to that server.

Clean Shutdown

The gateway handles SIGINT and SIGTERM by:

1. Stopping the stdin reader (using asyncio.StreamReader for clean cancellation)

2. Stopping all downstream subprocesses (SIGTERM with 5s grace, then SIGKILL)

3. Exiting with code 0

Request Timeouts

Each server has a configurable timeout (default 30s). If a downstream server doesn't respond within the timeout, the request fails with a StdioTransportError and the client receives a JSON-RPC error response. Per-request timeouts can also override the config default.

Docker

# Build the image
docker build -t mcp-gateway .

# Run with the default config (reads from stdin)
docker run --rm -i mcp-gateway

# Or mount your own config
docker run --rm -i -v /path/to/your_config.yaml:/app/gateway_config.yaml mcp-gateway

Testing

# Run all tests
python -m pytest tests/ -v

# Lint
ruff check mcp_gateway/ tests/ examples/ scripts/

104 tests across 7 files:

Project Structure

mcp-gw/
├── .github/
│   └── workflows/
│       └── ci.yml             # GitHub Actions: lint + test on push/PR
├── mcp_gateway/
│   ├── config/
│   │   ├── models.py          # Pydantic config models (with resiliency + learning fields)
│   │   └── loader.py          # YAML parser + validation
│   ├── transports/
│   │   └── stdio.py           # StdioTransport: subprocess bridge + crash monitoring
│   ├── registry/
│   │   └── tool_registry.py   # Multi-server registry with namespace routing
│   ├── gateway/
│   │   └── server.py          # Gateway: MCP server loop + routing + crash handling
│   ├── educational/
│   │   └── inspector.py       # ProtocolInspector: educational stderr traces
│   └── logging_utils.py       # Structured JSON logging
├── examples/
│   ├── mock_server.py               # Mock MCP server (default + alt tool sets)
│   ├── slow_server.py               # Slow MCP server for timeout testing
│   └── learning_network_config.yaml  # Local + live search config (Phase 6)
├── tests/
│   ├── conftest.py            # Shared fixtures (paths, config helpers)
│   ├── test_config_loader.py
│   ├── test_stdio_transport.py
│   ├── test_end_to_end.py
│   ├── test_registry.py
│   ├── test_multi_server.py
│   ├── test_resiliency.py
│   └── test_edge_cases.py
├── scripts/
│   ├── quick_e2e.py            # Single-server verification
│   ├── verify_multi_server.py  # Multi-server routing verification
│   ├── example_client.py      # Example MCP client connecting to the gateway
│   └── learn_traffic_init.py   # Educational traffic initiation (offline/online)
├── .gitignore
├── Dockerfile                  # Containerized deployment
├── gateway_config.yaml         # Default configuration (two mock servers)
├── LICENSE
└── pyproject.toml

Contributing

1. Fork the repo and clone your fork

2. Create a venv and install dev dependencies: pip install -e ".[dev]"

3. Make your changes

4. Run lint: ruff check mcp_gateway/ tests/ examples/ scripts/

5. Run tests: python -m pytest tests/ -v

6. Add tests for any new functionality

7. Submit a pull request

Build Status

Phase 1: Subprocess Stdio Bridge & Config Loader

• Project structure with gateway/, registry/, transports/, config/, tests/

• gateway_config.yaml parser using Pydantic for validation

• StdioTransport adapter: spawn subprocess, proxy JSON-RPC, initialize handshake

• End-to-end tool discovery (tools/list) and execution (tools/call) with mock server

• Structured JSON logging for all requests and responses

• Unit and integration tests with pytest

Phase 2: Multi-Server Registry & Namespace Router

• Extended registry to load multiple stdio servers from YAML config

• Namespace prefixing for tool schemas (server__tool)

• Automatic request routing back to the correct subprocess based on prefix

• Gateway acts as MCP server on stdin/stdout for AI clients

• Namespace collision handling (same tool name on different servers)

• Full test coverage for registry routing and multi-server integration

Phase 3: Lifecycle Management & Resiliency

• Process monitoring: detect unexpected subprocess crashes, reject in-flight requests

• Auto-restart: configurable per server, with max restart limit and tool re-discovery

• Clean shutdown on SIGINT/SIGTERM using asyncio.StreamReader for clean stdin cancellation

• Configurable per-server request timeouts (with per-request override)

• Full test coverage for crash recovery, restart, signal handling, and timeouts

Phase 4: Testing & Example Sandbox

• 104 tests across 7 files (config, transport, registry, multi-server, resiliency, edge cases)

• Shared test fixtures in conftest.py

• Example client script demonstrating real gateway usage

• Dockerfile for containerized deployment

• GitHub Actions CI (lint + test on push/PR)

• Ruff linting and formatting

• .gitignore, LICENSE, example configs

Phase 5: Educational Harness & Visual Inspection

• ProtocolInspector module with tree-formatted stderr traces

• LearningConfig with learning_mode and verbose_handshake fields

• --learn CLI flag to force-enable inspection from the command line

• Non-destructive: all output to stderr, stdout protocol stream untouched

• Global try/except wrapping for silent failure on unexpected errors

Phase 6: Live Endpoint Testing & Traffic Initiation

• learning_network_config.yaml with local mock + live Brave Search servers

• learn_traffic_init.py with auto-detection of BRAVE_API_KEY

• Offline fallback with educational text blocks (no stack traces)

• Packet journey visualization on stderr

• Graceful error handling for missing npx, invalid keys, or network failures

Roadmap

• Future: Resource proxying, prompt aggregation, SSE transport, persistence layer

Educational Harness & Visual Inspection

The gateway includes an optional educational mode that prints richly formatted, tree-like traces of all JSON-RPC traffic to stderr. This lets students and developers observe the MCP protocol in real time — initialize handshakes, tools/list aggregation, tools/call routing, namespace resolution, and downstream server communication — without affecting the protocol stream on stdout.

All inspection output goes exclusively to sys.stderr. The gateway's sys.stdout remains a clean JSON-RPC pipe, so AI clients (Claude Desktop, Cursor, etc.) are never disrupted.

Configuration

Add a learning block to gateway_config.yaml:

learning:
  learning_mode: true        # Enable protocol inspection (default: false)
  verbose_handshake: true    # Show detailed initialize breakdown (default: true)

If the learning block is omitted entirely, learning mode defaults to false and the gateway runs in its raw, high-performance proxy mode with zero overhead.

CLI Flag

Two ways of running mock with default and enhanced mode. Force-enable learning mode from the command line, regardless of config:

mcp-gateway gateway_config.yaml --learn
mcp-gateway enhanced_config.yaml --learn

Tracing During an Active Session

To watch protocol traffic while Claude Desktop (or any MCP host) is connected, redirect stderr to a log file and tail it in a separate terminal:

# Terminal 1: Run the gateway, pipe stderr to a trace file
mcp-gateway gateway_config.yaml --learn 2> trace.log

# Terminal 2: Watch the trace in real time
tail -f trace.log

Or in your Claude Desktop config, set the gateway command with stderr redirected:

{
  "mcpServers": {
    "gateway": {
      "command": "/path/to/mcp-gw/.venv/bin/mcp-gateway",
      "args": ["/path/to/mcp-gw/gateway_config.yaml", "--learn", "2>", "/tmp/mcp-trace.log"]
    }
  }
}

Then tail -f /tmp/mcp-trace.log in a terminal to see every request, response, routing decision, and error as it happens.

Live Endpoint Testing & Traffic Initiation

Traffic initiation is the act of programmatically generating MCP protocol requests to exercise the full gateway pipeline — from client intent, through namespace routing, to downstream server execution and back. The traffic init utility (scripts/learn_traffic_init.py) serves as a mock client that triggers this flow and prints a visual map of every step to stderr.

Packet Journey

When the traffic init script calls a tool, the request traverses the entire proxy architecture:

Client (learn_traffic_init.py)
  │  tools/call via stdin (JSON-RPC 2.0)
  ▼
Gateway (mcp-gateway)
  │  resolve namespace → identify server
  │  forward to subprocess stdin
  ▼
Downstream server (local or search)
  │  execute tool
  │  return result via stdout
  ▼
Gateway → Client (result on stdout)

Configuration

The learning network config (examples/learning_network_config.yaml) defines two server namespaces:

servers:
  - name: local        # Offline mock server (echo, add)
    command: python3
    args: [examples/mock_server.py]
    enabled: true

  - name: search       # Live Brave Search (requires BRAVE_API_KEY)
    command: npx
    args: ["-y", "@modelcontextprotocol/server-brave-search"]
    enabled: false     # Enabled automatically by the traffic init script

Offline Mode (no API key)

.venv/bin/python scripts/learn_traffic_init.py

Without BRAVE_API_KEY, the script targets the local mock server (local__echo) and prints an educational guide explaining how to obtain and configure the Brave Search API key for the live path.

Online Mode (with API key)

export BRAVE_API_KEY="your-api-key"
python scripts/learn_traffic_init.py

With the API key detected, the script enables the search server in the config, calls the live search__brave_web_search tool, and maps the full packet journey through the external endpoint. If the live server fails to start (missing npx, invalid key, network error), the script catches the error, prints an explanation, and falls back to offline mode — no stack trace.

License

MIT