MCP Servers in Production: Authentication, Discovery, and Scale

Building an MCP server is straightforward. Deploying one to production — where it handles real user requests, survives server failures, and scales to thousands of concurrent clients — is a different challenge entirely. The Model Context Protocol (MCP) defines how clients and servers communicate, but it does not dictate how to handle authentication, service discovery, or horizontal scaling. That is up to you.

In this post, we cover the three pillars of production MCP deployments: authentication and authorization, service discovery for dynamic environments, and scaling patterns that keep your servers responsive under load.

Why This Matters

MCP servers in development typically run as local processes over stdio transport. In production, they run as remote services over Streamable HTTP, serving many clients simultaneously. This shift introduces problems that stdio servers never face:

• Authentication: How do you verify that a client is authorized to use your tools?
• Discovery: How do clients find and connect to the right server in a dynamic environment?
• Scaling: How do you handle thousands of concurrent tool calls without degrading performance?

Getting these wrong means your MCP server is a development toy, not a production service.

Authentication and Authorization

MCP recommends OAuth 2.1 for authentication on the Streamable HTTP transport. Here is how to implement it properly:

OAuth 2.1 With the MCP SDK

from mcp.server import MCPServer
from mcp.server.auth import OAuthServerProvider, BearerTokenAuth

class CustomOAuthProvider(OAuthServerProvider):
    def __init__(self, user_store: dict, token_store: dict):
        self.user_store = user_store
        self.token_store = token_store

    async def verify_token(self, token: str) -> dict:
        """Verify an access token and return the associated claims."""
        token_data = self.token_store.get(token)
        if not token_data:
            raise ValueError("Invalid token")

        # Check expiration
        from datetime import datetime, timezone
        if datetime.now(timezone.utc) > token_data["expires_at"]:
            del self.token_store[token]
            raise ValueError("Token expired")

        return {
            "sub": token_data["user_id"],
            "scopes": token_data["scopes"],
            "client_id": token_data["client_id"],
        }

    async def authorize(self, client_id: str, scopes: list[str]) -> str:
        """Issue an authorization code."""
        import secrets
        auth_code = secrets.token_urlsafe(32)
        self.user_store[auth_code] = {
            "client_id": client_id,
            "scopes": scopes,
            "created_at": datetime.now(timezone.utc),
        }
        return auth_code

# Configure the server with OAuth
mcp = MCPServer(
    "production-server",
    auth=OAuthServerProvider(
        provider=CustomOAuthProvider(user_store={}, token_store={}),
        # Optional: JWT-based verification for distributed systems
        # jwt_issuer="https://auth.yourcompany.com",
        # jwt_audience="mcp-servers",
    ),
)

Scope-Based Access Control

Not every client should have access to every tool. Implement scope-based authorization:

from functools import wraps

def require_scope(scope: str):
    """Decorator to require a specific OAuth scope for a tool."""
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, context=None, **kwargs):
            if context and scope not in context.scopes:
                return {
                    "content": [{"type": "text", "text": f"Insufficient permissions. Required scope: {scope}"}],
                    "isError": True,
                }
            return await func(*args, context=context, **kwargs)
        return wrapper
    return decorator

@mcp.tool()
@require_scope("orders:read")
async def get_order(order_id: str) -> str:
    """Get order details. Requires orders:read scope."""
    # ... implementation ...

@mcp.tool()
@require_scope("orders:write")
async def process_refund(order_id: str, amount: float) -> str:
    """Process a refund. Requires orders:write scope."""
    # ... implementation ...

Token Refresh and Rotation

Production tokens should have short lifetimes and support refresh:

import secrets
from datetime import datetime, timezone, timedelta

class TokenManager:
    def __init__(self):
        self.tokens = {}

    def create_token(self, user_id: str, scopes: list[str], ttl_seconds: int = 3600) -> dict:
        token = secrets.token_urlsafe(48)
        refresh_token = secrets.token_urlsafe(48)

        self.tokens[token] = {
            "user_id": user_id,
            "scopes": scopes,
            "created_at": datetime.now(timezone.utc),
            "expires_at": datetime.now(timezone.utc) + timedelta(seconds=ttl_seconds),
            "refresh_token": refresh_token,
        }

        return {
            "access_token": token,
            "token_type": "Bearer",
            "expires_in": ttl_seconds,
            "refresh_token": refresh_token,
            "scope": " ".join(scopes),
        }

    def refresh(self, refresh_token: str) -> dict | None:
        for token, data in self.tokens.items():
            if data["refresh_token"] == refresh_token:
                # Invalidate old token
                del self.tokens[token]
                # Issue new token
                return self.create_token(data["user_id"], data["scopes"])
        return None

Service Discovery

In dynamic environments (Kubernetes, ECS, service meshes), MCP clients need to discover available servers. MCP supports this through the server/discover method and the notifications/tools/list_changed notification.

Building a Discovery Registry

from dataclasses import dataclass
import asyncio
import time

@dataclass
class MCPServerInfo:
    name: str
    url: str
    capabilities: dict
    last_heartbeat: float
    status: str = "healthy"

class MCPDiscoveryRegistry:
    def __init__(self):
        self.servers: dict[str, MCPServerInfo] = {}
        self.lock = asyncio.Lock()

    async def register(self, server: MCPServerInfo):
        """Register an MCP server."""
        async with self.lock:
            server.last_heartbeat = time.time()
            self.servers[server.name] = server

    async def heartbeat(self, server_name: str):
        """Update heartbeat for a server."""
        async with self.lock:
            if server_name in self.servers:
                self.servers[server_name].last_heartbeat = time.time()

    async def discover(self, capability_filter: str = None) -> list[MCPServerInfo]:
        """Discover available servers, optionally filtered by capability."""
        async with self.lock:
            now = time.time()
            healthy = []
            for server in self.servers.values():
                # Mark servers with stale heartbeats as unhealthy
                if now - server.last_heartbeat > 60:
                    server.status = "unhealthy"
                    continue

                if capability_filter:
                    if capability_filter in server.capabilities:
                        healthy.append(server)
                else:
                    healthy.append(server)

            return healthy

    async def get_tools(self) -> list[dict]:
        """Get all tools from all healthy servers."""
        tools = []
        servers = await self.discover()
        for server in servers:
            # In production, this would call server/discover and tools/list
            tools.extend(server.capabilities.get("tools", []))
        return tools

# Usage
registry = MCPDiscoveryRegistry()

# Servers register themselves on startup
await registry.register(MCPServerInfo(
    name="order-service",
    url="https://orders.internal/mcp",
    capabilities={"tools": ["get_order", "process_refund"]},
    last_heartbeat=time.time(),
))

# Clients discover servers
servers = await registry.discover(capability_filter="orders")

Health Check Pattern

MCP servers should expose health endpoints for load balancers:

from fastapi import FastAPI
from contextlib import asynccontextmanager
import asyncio

app = FastAPI()

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Start heartbeat loop
    heartbeat_task = asyncio.create_task(send_heartbeats())
    yield
    heartbeat_task.cancel()

async def send_heartbeats():
    """Send periodic heartbeats to the discovery registry."""
    while True:
        try:
            async with httpx.AsyncClient() as client:
                await client.post(
                    "https://discovery.internal/heartbeat",
                    json={"server": "order-service", "status": "healthy"},
                )
        except Exception:
            pass
        await asyncio.sleep(30)

@app.get("/health")
async def health_check():
    return {"status": "healthy", "server": "order-service"}

Scaling MCP Servers

Horizontal Scaling With Load Balancing

For Streamable HTTP servers, use a load balancer to distribute requests:

# Kubernetes deployment
apiVersion: apps/v1
kind: Deployment
metadata:
  name: mcp-order-service
spec:
  replicas: 3
  selector:
    matchLabels:
      app: mcp-order-service
  template:
    metadata:
      labels:
        app: mcp-order-service
    spec:
      containers:
      - name: server
        image: yourregistry/mcp-order-service:latest
        ports:
        - containerPort: 8000
        env:
        - name: MCP_TRANSPORT
          value: "streamable-http"
        - name: MCP_HOST
          value: "0.0.0.0"
        - name: MCP_PORT
          value: "8000"
        readinessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 5
          periodSeconds: 10
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 10
          periodSeconds: 30
        resources:
          requests:
            memory: "256Mi"
            cpu: "250m"
          limits:
            memory: "512Mi"
            cpu: "500m"
---
apiVersion: v1
kind: Service
metadata:
  name: mcp-order-service
spec:
  selector:
    app: mcp-order-service
  ports:
  - port: 80
    targetPort: 8000
  type: ClusterIP

Connection Pooling

For servers that call downstream APIs, implement connection pooling:

import httpx
from contextlib import asynccontextmanager

class MCPConnectionPool:
    def __init__(self, max_connections: int = 20):
        self.max_connections = max_connections
        self.pool = None

    async def __aenter__(self):
        self.pool = httpx.AsyncClient(
            limits=httpx.Limits(
                max_connections=self.max_connections,
                max_keepalive_connections=10,
                keepalive_expiry=30,
            ),
            timeout=httpx.Timeout(30.0),
        )
        return self.pool

    async def __aexit__(self, *args):
        await self.pool.aclose()

# Usage in tools
@mcp.tool()
async def get_order(order_id: str) -> str:
    async with MCPConnectionPool() as client:
        response = await client.get(f"{ORDER_API_BASE}/{order_id}")
        return response.json()

Caching Tool Responses

Cache responses for tools that return stable data:

from functools import lru_cache
from datetime import datetime, timedelta
import asyncio

class ToolCache:
    def __init__(self, default_ttl: int = 300):
        self.cache = {}
        self.default_ttl = default_ttl

    def get(self, key: str):
        if key in self.cache:
            value, expiry = self.cache[key]
            if datetime.now() < expiry:
                return value
            del self.cache[key]
        return None

    def set(self, key: str, value, ttl: int = None):
        expiry = datetime.now() + timedelta(seconds=ttl or self.default_ttl)
        self.cache[key] = (value, expiry)

tool_cache = ToolCache(default_ttl=60)

@mcp.tool()
async def get_policy(policy_name: str) -> str:
    cached = tool_cache.get(f"policy:{policy_name}")
    if cached:
        return cached

    # Fetch from source
    result = await fetch_policy(policy_name)
    tool_cache.set(f"policy:{policy_name}", result, ttl=600)
    return result

Best Practices

• Use short-lived tokens: Access tokens should expire within 1 hour. Implement refresh token rotation for long-lived sessions.
• Implement rate limiting: Apply per-client rate limits to prevent abuse. Use token bucket or sliding window algorithms.
• Monitor everything: Track tool call latency, error rates, and cache hit rates. Export these metrics via OpenTelemetry.
• Version your tools: Use semantic versioning for MCP server releases. When tools change, send notifications/tools/list_changed so clients refresh their tool registry.
• Plan for failures: Every tool should have a timeout, a retry policy, and a circuit breaker. A slow database should not block your MCP server.

Common Pitfalls

• Hardcoded credentials: Never embed API keys or tokens in server code. Use environment variables or a secrets manager.
• Ignoring transport security: Always use HTTPS for Streamable HTTP. STDIO servers should only run locally.
• Single-server architecture: A single MCP server is a single point of failure. Deploy at least two replicas behind a load balancer.
• Skipping the discovery handshake: Always implement server/discover. Clients that skip discovery cannot detect capability changes.

Conclusion and Next Steps

Production MCP servers need authentication, discovery, and scaling. Start with OAuth 2.1 for authentication, a registry for discovery, and horizontal replicas for scaling. Monitor your servers, cache aggressively, and always have a health check.

The next step is implementing the MCP Tasks extension for long-running tool calls — essential for operations like report generation or data processing that take more than a few seconds.