Building Custom MCP Servers for Enterprise

The Model Context Protocol (MCP) is an open standard that defines how agents discover and call tools exposed by external servers. Rather than hardcoding tool logic into each agent, you build MCP servers that expose enterprise capabilities, database queries, internal APIs, document retrieval, workflow triggers, as standardized, discoverable tools. Any agent that speaks MCP can immediately use any MCP server. This lesson builds a production-grade enterprise MCP server: authenticated, versioned, containerized, and observable.

MCP Protocol Architecture

MCP uses JSON-RPC 2.0 as its transport layer. Servers expose three primitive types: tools (callable functions with typed input schemas), resources (readable data sources like documents or database records), and prompts (reusable prompt templates with parameters). Agents connect to MCP servers at session start, call tools/list to discover available capabilities, and call tools/call to invoke them. The protocol supports both stdio (for local subprocess servers) and HTTP/SSE (for networked servers, which is the enterprise pattern).

flowchart TD
    A1["Research Agent"] --> D["Agent Discovery\ntools/list"]
    A2["Writer Agent"] --> D
    A3["Compliance Agent"] --> D
    D --> MCP["Enterprise MCP Server\n(HTTP + SSE)"]
    MCP --> T1["Tool: crm_search/v2"]
    MCP --> T2["Tool: kb_query/v1"]
    MCP --> T3["Tool: workflow_trigger/v1"]
    T1 --> SF["Salesforce API"]
    T2 --> KB["Knowledge Base\nQdrant"]
    T3 --> WF["Workflow Engine\n(internal API)"]

    style A1 fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style A2 fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style A3 fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style D fill:#fff7ed,stroke:#f59e0b,color:#0F172A
    style MCP fill:#f0fdf9,stroke:#0D9488,color:#0F172A
    style T1 fill:#f0fdf9,stroke:#0D9488,color:#0F172A
    style T2 fill:#f0fdf9,stroke:#0D9488,color:#0F172A
    style T3 fill:#f0fdf9,stroke:#0D9488,color:#0F172A
    style SF fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style KB fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style WF fill:#EEF0F7,stroke:#6366F1,color:#0F172A

Building an MCP Server with the Python SDK

The mcp Python SDK provides a FastMCP class that handles protocol negotiation, tool registration, and JSON-RPC serialization. You define tools as decorated async functions with Pydantic-validated inputs.

from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel, Field
import asyncpg

# FastMCP handles JSON-RPC 2.0 transport, capability negotiation,
# and error formatting: you write pure async Python functions.
mcp = FastMCP(
    name="enterprise-knowledge-server",
    version="1.2.0",
    description="Internal knowledge base and CRM search capabilities",
)

class KBSearchInput(BaseModel):
    query: str = Field(..., min_length=3, max_length=500, description="Search query text")
    top_k: int = Field(5, ge=1, le=20, description="Number of results to return")
    doc_type: str | None = Field(None, description="Filter by document type: policy, procedure, faq")

class KBSearchResult(BaseModel):
    doc_id: str
    title: str
    excerpt: str
    relevance_score: float
    doc_type: str

@mcp.tool(name="kb_query/v1", description="Search the internal knowledge base by semantic similarity")
async def kb_query(input: KBSearchInput) -> list[KBSearchResult]:
    """
    Semantic search over the internal knowledge base.
    We validate input with Pydantic BEFORE touching the database
    invalid inputs are rejected at the protocol layer with a structured
    error, not a 500 from the database driver.
    """
    pool = mcp.state["db_pool"]  # Injected at startup, shared across requests
    embedding = await generate_embedding(input.query)  # Your embedding function

    type_filter = "AND doc_type = $3" if input.doc_type else ""
    params = [embedding, input.top_k, input.doc_type] if input.doc_type else [embedding, input.top_k]

    rows = await pool.fetch(
        f"""
        SELECT doc_id, title, excerpt, doc_type,
               1 - (embedding <=> $1::vector) AS relevance_score
        FROM knowledge_base
        WHERE 1=1 {type_filter}
        ORDER BY embedding <=> $1::vector
        LIMIT $2
        """,
        *params,
    )

    return [KBSearchResult(**dict(row)) for row in rows]

Authentication via HTTP Headers

Enterprise MCP servers must authenticate callers. For internal agent networks, API key authentication via a custom HTTP header is the simplest pattern that works across all agent frameworks. Validate the key against a secrets store on every request, not on startup, so key rotation takes effect immediately without a server restart.

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
import hashlib, hmac

VALID_KEY_HASHES: set[str] = set()  # Pre-loaded from secrets manager at startup

class APIKeyMiddleware(BaseHTTPMiddleware):
    """
    Validate X-Agent-API-Key header on every MCP request.
    We compare hashes, not raw keys, this prevents timing attacks
    and means we never need to store the plaintext key in memory.
    """
    async def dispatch(self, request: Request, call_next):
        if request.url.path in ("/health", "/ready"):
            # Health probes do not require authentication
            # Kubernetes/load balancers need these to work pre-auth
            return await call_next(request)

        api_key = request.headers.get("X-Agent-API-Key", "")
        key_hash = hashlib.sha256(api_key.encode()).hexdigest()

        if key_hash not in VALID_KEY_HASHES:
            return JSONResponse(
                status_code=401,
                content={"error": "Invalid or missing API key"},
            )

        return await call_next(request)

Versioning MCP Tools with Semantic Versioning

Tool versioning in MCP is done by including the version in the tool name: kb_query/v1, kb_query/v2. This is the simplest approach that works with all MCP clients, no custom capability negotiation required. Register both versions simultaneously during migration windows.

sequenceDiagram
    participant OLD as Old Agent (v1 client)
    participant NEW as New Agent (v2 client)
    participant MCP as MCP Server

    OLD->>MCP: tools/list
    MCP-->>OLD: [kb_query/v1, kb_query/v2, crm_search/v1]
    OLD->>MCP: tools/call kb_query/v1 (works as before)
    MCP-->>OLD: Results

    NEW->>MCP: tools/list
    MCP-->>NEW: [kb_query/v1, kb_query/v2, crm_search/v1]
    NEW->>MCP: tools/call kb_query/v2 (uses new schema)
    MCP-->>NEW: Enhanced results with new fields

    Note over MCP: v1 deprecated header in response
    Note over MCP: Remove v1 once all agents migrate

    style OLD fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style NEW fill:#f0fdf9,stroke:#0D9488,color:#0F172A
    style MCP fill:#fff7ed,stroke:#f59e0b,color:#0F172A

Deploying as a Docker Container

MCP servers should be containerized and deployed behind an internal load balancer. They are stateless (all state lives in the databases they connect to), so horizontal scaling is trivial. The Dockerfile is minimal: a Python base image, dependency installation, and an entrypoint that starts the server with uvicorn.

# Startup: inject shared resources into MCP server state
import asyncpg, os
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app):
    """
    Create shared connection pool at startup, not per-request.
    Connection pools are expensive to create, one pool per process,
    shared across all concurrent tool calls via asyncpg's pool.acquire().
    """
    pool = await asyncpg.create_pool(
        dsn=os.environ["DATABASE_URL"],
        min_size=5,
        max_size=20,
        command_timeout=30,
    )
    mcp.state["db_pool"] = pool
    yield
    await pool.close()

# Start the server: uvicorn enterprise_mcp:app --host 0.0.0.0 --port 8080
app = mcp.streamable_http_app(lifespan=lifespan)

Agent Discovery and MCP Server Health Monitoring

In a production network, agents discover MCP servers via a central registry: a simple key-value store (Redis or Consul) where each MCP server registers its address, name, version, and health status at startup. Agents query the registry on initialization to build their tool catalog. Each MCP server exposes a /health endpoint that the registry polls every 30 seconds; unhealthy servers are removed from the registry automatically, and agents that cache the tool list re-query after a cache miss or error.

Monitor MCP servers like any microservice: track tool call latency (p50/p95/p99), error rate by tool name and version, and authentication failure rate (a spike signals a credential rotation issue or a misconfigured agent). Export these metrics to your existing observability stack via OpenTelemetry, the MCP server is just an HTTP service from the infrastructure’s perspective.

This lesson completes the module on enterprise integrations. You now have the full stack: distributed state that is safe under concurrent agent writes, long-term memory with semantic retrieval, typed clients for Salesforce and SAP, database agents with cost controls, and custom MCP servers that expose all of these capabilities as standardized, discoverable tools for your entire agent network.