handles image caching, container creation with security constraints, and health monitoring.
3. Bridge Layer: A protocol translator that wraps stdio-based MCP servers and exposes them as HTTP endpoints. This allows HTTP-based AI clients to communicate with subprocess-based MCP servers.
Implementation Details
1. The MCP Bridge: Protocol Translation
Most MCP servers are designed to run as subprocesses communicating via standard input/output (stdio). The Bridge component resolves the mismatch between stdio servers and HTTP clients. It launches the MCP server as a subprocess, performs the initialization handshake, and exposes HTTP endpoints for tool listing and invocation.
The Bridge implementation must handle subprocess lifecycle management and error propagation. If a server fails to start due to missing credentials, the Bridge must detect the premature closure of stdout and report a clear error rather than hanging.
import asyncio
import shlex
import os
from typing import List, Dict, Any
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
class ToolCallRequest(BaseModel):
tool: str
arguments: Dict[str, Any]
class McpBridge:
def __init__(self, command: str):
self.command = shlex.split(command)
self.process: asyncio.subprocess.Process | None = None
self.initialized = False
async def initialize(self):
"""Launch subprocess and perform MCP handshake."""
self.process = await asyncio.create_subprocess_exec(
*self.command,
stdin=asyncio.subprocess.PIPE,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
env={**os.environ, "MCP_TRANSPORT": "stdio"}
)
# Perform MCP initialize handshake
init_payload = {
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {"protocolVersion": "2024-11-05"}
}
await self.process.stdin.write(
(json.dumps(init_payload) + "\n").encode()
)
await self.process.stdin.drain()
# Read response; if process exits, credentials are likely missing
response = await self._read_response()
if response is None:
raise RuntimeError("MCP server failed to initialize. Check credentials.")
self.initialized = True
async def list_tools(self) -> List[Dict]:
if not self.initialized:
raise HTTPException(status_code=503, detail="Bridge not initialized")
payload = {"jsonrpc": "2.0", "id": 2, "method": "tools/list"}
await self.process.stdin.write((json.dumps(payload) + "\n").encode())
await self.process.stdin.drain()
result = await self._read_response()
return result.get("result", [])
async def call_tool(self, req: ToolCallRequest) -> Dict:
if not self.initialized:
raise HTTPException(status_code=503, detail="Bridge not initialized")
payload = {
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
"params": {"name": req.tool, "arguments": req.arguments}
}
await self.process.stdin.write((json.dumps(payload) + "\n").encode())
await self.process.stdin.drain()
return await self._read_response()
async def _read_response(self) -> Dict | None:
"""Read JSON-RPC response from stdout."""
try:
line = await asyncio.wait_for(self.process.stdout.readline(), timeout=10.0)
if not line:
return None
return json.loads(line.decode().strip())
except asyncio.TimeoutError:
return None
2. The Orchestrator: Secure Container Management
The Orchestrator is responsible for provisioning containers with strict security constraints. It must ensure that containers run with minimal privileges, are isolated from the host network, and have resource limits enforced.
Key security measures include dropping all Linux capabilities, preventing privilege escalation, running as a non-root user, and attaching containers to a dedicated Docker network.
import docker
from docker.types import Resources, Ulimit
class ContainerOrchestrator:
def __init__(self, docker_client: docker.DockerClient):
self.client = docker_client
self.network_name = "mcp_isolated_net"
def _ensure_image(self, image_name: str):
"""Optimization: Skip pull if image exists locally."""
try:
self.client.images.get(image_name)
return True
except docker.errors.ImageNotFound:
self.client.images.pull(image_name)
return False
def deploy_bridge(self, image: str, mcp_command: str, env_vars: Dict[str, str]) -> str:
"""Deploy a hardened MCP Bridge container."""
self._ensure_image(image)
security_opts = {
"cap_drop": ["ALL"],
"no_new_privileges": True,
"user": "bridge"
}
resource_limits = Resources(
cpu_shares=512,
mem_limit="512m",
memswap_limit="512m"
)
container = self.client.containers.run(
image=image,
detach=True,
name=f"mcp-bridge-{uuid.uuid4().hex[:8]}",
environment={
"MCP_COMMAND": mcp_command,
**env_vars
},
security_opt=security_opts,
resources=resource_limits,
network=self.network_name,
restart_policy={"Name": "no"}
)
return container.id
3. Credential Management and Validation
Credentials must never be stored in the database or logged. The system uses a declarative approach where each allowed MCP server defines its required environment variables. Before deployment, the frontend collects these values via a secure modal and passes them directly to the Orchestrator in the deployment request. The Orchestrator injects them as environment variables into the container. Once deployed, the credentials are inaccessible to developers and are excluded from all log outputs.
4. Audit and Access Control
The Gateway layer intercepts all requests. It validates the user's bearer token, verifies that the requested tool is in the allowlist for that user's role, and logs the invocation. Logs include the member ID, server slug, tool name, latency, and HTTP status. Inputs and outputs are not stored to maintain data privacy and reduce storage costs. This design ensures GDPR compliance while providing a complete audit trail for security reviews.
Pitfall Guide
Implementing enterprise MCP infrastructure introduces specific technical challenges. The following pitfalls are common in production environments and include mitigation strategies.
| Pitfall | Explanation | Fix |
|---|
| Silent Handshake Failure | MCP servers that require credentials may exit immediately if environment variables are missing. The Bridge process closes stdout before the handshake completes, resulting in a generic error. | Implement a pre-deploy validation step. Define required_env_vars for each server and enforce collection via a UI modal before sending the deploy request. |
| Image Pull Latency | Attempting to pull a local-only image from a registry on every deployment causes failures and adds latency. | Check for the image locally using docker.images.get() before attempting a pull. Only pull if the image is missing. |
| Privilege Escalation | Malicious or compromised MCP packages may attempt to use setuid binaries or file capabilities to gain root access. | Apply no-new-privileges: true and cap_drop: ALL to the container configuration. Run the container as a non-root user. |
| Credential Leakage | Environment variables containing secrets may accidentally appear in container logs or error messages. | Configure the logging system to redact known secret patterns. Never log environment variables. Ensure the Bridge does not echo command arguments that contain tokens. |
| Resource Exhaustion | A runaway MCP server process can consume excessive CPU or memory, affecting other containers on the host. | Enforce strict resource limits (cpu_shares, mem_limit) per container. Use resource profiles (small/medium/large) based on the expected workload of the server. |
| Network Bleed | Containers may inadvertently access the host network or other internal services. | Attach containers to a dedicated Docker bridge network with no external routing. Disable host network mode. |
| Cold Start Delays | Installing MCP server packages via npm or pip on every container start increases deployment time. | Pre-install common MCP server packages in the Bridge base image. Use a multi-stage build to cache dependencies. |
Production Bundle
Action Checklist
Decision Matrix
| Scenario | Recommended Approach | Why | Cost Impact |
|---|
| Individual Developer | Local npx execution | Fastest setup; no infrastructure overhead | Low |
| Small Team (<10 devs) | Shared Orchestrator with Gateway | Centralized control; reduced credential risk | Medium |
| Regulated Enterprise | Containerized Gateway + Dedicated Orchestrator | Full audit trail; strict isolation; compliance | High |
| High-Volume Production | Dedicated Orchestrator per region | Reduced latency; fault isolation | High |
| On-Premise Requirement | Self-hosted Docker deployment | Data sovereignty; no external dependencies | Medium |
Configuration Template
The following Docker Compose configuration demonstrates a self-hosted Bridge node with security hardening. This template can be adapted for use in a Kubernetes deployment or as part of a larger orchestration stack.
version: '3.8'
services:
mcp-bridge:
image: mcp-bridge:latest
environment:
- MCP_COMMAND=npx -y @modelcontextprotocol/server-filesystem /workspace
- GITHUB_TOKEN=${GITHUB_TOKEN}
security_opt:
- no-new-privileges:true
cap_drop:
- ALL
user: "bridge"
networks:
- mcp_isolated
deploy:
resources:
limits:
cpus: '0.50'
memory: 512M
reservations:
cpus: '0.25'
memory: 256M
restart: "no"
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:8080/health"]
interval: 30s
timeout: 10s
retries: 3
networks:
mcp_isolated:
driver: bridge
internal: true
Quick Start Guide
- Deploy the Orchestrator: Provision a server with Docker installed and run the Orchestrator service. Configure it to connect to your Docker socket and set up the isolated network.
- Configure the Gateway: Deploy the Gateway service and configure it with your authentication provider. Define the allowlists for your team members.
- Create the Bridge Image: Build the Bridge Docker image with the required MCP server packages. Tag and load the image onto the Orchestrator host.
- Define a Server: Add an entry to your server catalog with the command and required environment variables.
- Deploy and Test: Use the dashboard to deploy a Bridge container. Verify that the container starts, the health check passes, and the Gateway can proxy requests to the tool.
