Model Context Protocol — Building and Deploying MCP Servers for Production AI Agents

What MCP Is and Why It Exists

The Model Context Protocol (MCP) is an open standard published by Anthropic in late 2024 that defines how AI models communicate with external tools, data sources, and execution environments. Before MCP, every team that wanted to connect an LLM to a database, an API, or a code interpreter had to invent their own integration layer: custom function schemas, ad-hoc authentication, bespoke error handling. The result was a proliferation of incompatible “tool calling” implementations that could not be shared across models or frameworks.

MCP solves this by defining a common wire protocol between three roles: hosts (the LLM runtime — Claude, a local model, an agent framework), clients (the protocol adapter inside the host), and servers (the external capability providers — a GitHub integration, a SQL query engine, a file system accessor). Any MCP server works with any MCP-compatible host without modification. This is the same value proposition that made USB successful: one connector standard, infinite peripherals.

MCP servers expose three primitive capability types: Tools (callable functions the LLM can invoke, with a JSON Schema describing parameters), Resources (data the LLM can read — files, database records, API responses — identified by URI), and Prompts (reusable, parameterised prompt templates that the host can present to the user or inject into context). Together these three primitives cover the vast majority of what production AI agents need.

MCP Architecture — Host, Client, Server, Transport

Understanding the four-layer model is essential before writing a single line of server code. The layers have clearly separated responsibilities and knowing where each concern lives prevents you from building things in the wrong place.

Building a TypeScript MCP Server

The @modelcontextprotocol/sdk is the official TypeScript implementation. Install it alongside zod for runtime schema validation:

npm install @modelcontextprotocol/sdk zod
# or
pnpm add @modelcontextprotocol/sdk zod

The following example builds a server that exposes a query_database tool, a resource for reading schema metadata, and a prompt template for generating SQL from natural language:

// server.ts
import { McpServer, ResourceTemplate } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";
import { Pool } from "pg";

const pool = new Pool({ connectionString: process.env.DATABASE_URL });

const server = new McpServer({
  name: "postgres-mcp-server",
  version: "1.0.0",
});

// --- Tool: execute a read-only SQL query ---
server.tool(
  "query_database",
  "Execute a read-only SQL SELECT query and return results as JSON",
  {
    sql: z.string().describe("The SQL SELECT query to execute"),
    limit: z.number().int().min(1).max(1000).default(100)
      .describe("Maximum rows to return"),
  },
  async ({ sql, limit }) => {
    // Security: only allow SELECT statements
    const trimmed = sql.trim().toUpperCase();
    if (!trimmed.startsWith("SELECT")) {
      return {
        content: [{ type: "text", text: "Error: only SELECT queries are permitted" }],
        isError: true,
      };
    }

    const limitedSql = `${sql.trim().replace(/;+$/, "")} LIMIT ${limit}`;
    const result = await pool.query(limitedSql);

    return {
      content: [
        {
          type: "text",
          text: JSON.stringify({
            rows: result.rows,
            rowCount: result.rowCount,
            fields: result.fields.map((f) => ({ name: f.name, dataTypeID: f.dataTypeID })),
          }, null, 2),
        },
      ],
    };
  }
);

// --- Resource: expose table schema metadata ---
server.resource(
  "schema",
  new ResourceTemplate("schema://{tableName}", { list: undefined }),
  async (uri, { tableName }) => {
    const result = await pool.query(
      `SELECT column_name, data_type, is_nullable
       FROM information_schema.columns
       WHERE table_name = $1
       ORDER BY ordinal_position`,
      [tableName]
    );

    return {
      contents: [
        {
          uri: uri.href,
          text: JSON.stringify(result.rows, null, 2),
          mimeType: "application/json",
        },
      ],
    };
  }
);

// --- Prompt: natural language to SQL template ---
server.prompt(
  "nl_to_sql",
  "Generate a SQL SELECT query from a natural language description",
  {
    description: z.string().describe("Natural language description of the data you want"),
    table: z.string().describe("Target table name"),
  },
  ({ description, table }) => ({
    messages: [
      {
        role: "user",
        content: {
          type: "text",
          text: `Generate a read-only PostgreSQL SELECT query for the table "${table}".
Goal: ${description}
Return only the SQL query, no explanation.`,
        },
      },
    ],
  })
);

// --- Start the server over stdio ---
async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
  console.error("MCP server running on stdio");
}

main().catch((err) => {
  console.error("Fatal:", err);
  process.exit(1);
});

HTTP Streamable Transport (Cloud Deployment)

For deployments where the server is a remote HTTP service rather than a local subprocess, swap the transport to StreamableHTTPServerTransport:

// http-server.ts
import express from "express";
import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js";
import { randomUUID } from "crypto";

const app = express();
app.use(express.json());

// Session map — one transport instance per connected client session
const sessions = new Map<string, StreamableHTTPServerTransport>();

app.post("/mcp", async (req, res) => {
  const sessionId = req.headers["mcp-session-id"] as string | undefined;

  let transport: StreamableHTTPServerTransport;

  if (sessionId && sessions.has(sessionId)) {
    transport = sessions.get(sessionId)!;
  } else {
    // New session — create transport and connect the MCP server
    transport = new StreamableHTTPServerTransport({
      sessionIdGenerator: () => randomUUID(),
      onsessioninitialized: (id) => sessions.set(id, transport),
    });
    // buildMcpServer() returns your McpServer instance (same as above)
    const mcpServer = buildMcpServer();
    await mcpServer.connect(transport);
  }

  await transport.handleRequest(req, res, req.body);
});

app.delete("/mcp", async (req, res) => {
  const sessionId = req.headers["mcp-session-id"] as string | undefined;
  if (sessionId) {
    const transport = sessions.get(sessionId);
    if (transport) {
      await transport.close();
      sessions.delete(sessionId);
    }
  }
  res.status(200).end();
});

app.listen(3000, () => console.log("MCP HTTP server listening on :3000"));

Building a Python MCP Server

The official mcp Python package provides a high-level decorator API that maps closely to the TypeScript SDK. Install with uv or pip:

uv add mcp
# or
pip install mcp

# server.py
import asyncio
import json
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import (
    Tool, TextContent, Resource, Prompt, PromptMessage, PromptArgument,
    GetPromptResult,
)
import httpx

app = Server("weather-mcp-server")

# --- Tool: fetch weather data ---
@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="get_weather",
            description="Fetch current weather and 7-day forecast for a city",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "City name, e.g. 'Warsaw' or 'New York'",
                    },
                    "units": {
                        "type": "string",
                        "enum": ["metric", "imperial"],
                        "default": "metric",
                    },
                },
                "required": ["city"],
            },
        )
    ]


@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name != "get_weather":
        raise ValueError(f"Unknown tool: {name}")

    city = arguments["city"]
    units = arguments.get("units", "metric")

    async with httpx.AsyncClient(timeout=10.0) as client:
        resp = await client.get(
            "https://wttr.in/{city}",
            params={"format": "j1", "lang": "en"},
            headers={"User-Agent": "mcp-weather-server/1.0"},
        )
        resp.raise_for_status()
        data = resp.json()

    # Extract current conditions and 3-day forecast
    current = data["current_condition"][0]
    result = {
        "city": city,
        "temperature_c": current["temp_C"],
        "temperature_f": current["temp_F"],
        "description": current["weatherDesc"][0]["value"],
        "humidity_pct": current["humidity"],
        "wind_kmh": current["windspeedKmph"],
        "forecast_3day": [
            {
                "date": day["date"],
                "max_c": day["maxtempC"],
                "min_c": day["mintempC"],
                "description": day["hourly"][4]["weatherDesc"][0]["value"],
            }
            for day in data["weather"][:3]
        ],
    }

    return [TextContent(type="text", text=json.dumps(result, indent=2))]


# --- Resource: expose a curated city list ---
@app.list_resources()
async def list_resources() -> list[Resource]:
    return [
        Resource(
            uri="config://supported-cities",
            name="Supported Cities",
            description="List of cities with reliable weather data coverage",
            mimeType="application/json",
        )
    ]


@app.read_resource()
async def read_resource(uri: str) -> str:
    if uri == "config://supported-cities":
        cities = ["Warsaw", "Berlin", "London", "New York", "Tokyo", "Sydney"]
        return json.dumps({"cities": cities})
    raise ValueError(f"Unknown resource: {uri}")


# --- Prompt: weather briefing template ---
@app.list_prompts()
async def list_prompts() -> list[Prompt]:
    return [
        Prompt(
            name="weather_briefing",
            description="Generate a concise weather briefing for a travel itinerary",
            arguments=[
                PromptArgument(name="city", description="Destination city", required=True),
                PromptArgument(name="arrival_date", description="ISO 8601 arrival date", required=True),
            ],
        )
    ]


@app.get_prompt()
async def get_prompt(name: str, arguments: dict) -> GetPromptResult:
    if name != "weather_briefing":
        raise ValueError(f"Unknown prompt: {name}")

    city = arguments["city"]
    date = arguments["arrival_date"]

    return GetPromptResult(
        description=f"Weather briefing for {city} on {date}",
        messages=[
            PromptMessage(
                role="user",
                content=TextContent(
                    type="text",
                    text=(
                        f"I'm travelling to {city} arriving on {date}. "
                        f"Please use the get_weather tool to fetch current conditions, "
                        f"then write a 3-sentence travel weather briefing covering "
                        f"what to expect and what to pack."
                    ),
                ),
            )
        ],
    )


async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(read_stream, write_stream, app.create_initialization_options())


if __name__ == "__main__":
    asyncio.run(main())

Authentication and Authorization Patterns

MCP 2025-03-26 formally adopts OAuth 2.1 as the standard authentication mechanism for remote servers. The protocol defines a metadata discovery endpoint (/.well-known/oauth-authorization-server) and a mandatory PKCE flow for all clients. For internal or self-hosted deployments, API key authentication via a custom header is simpler and sufficient.

API Key Middleware (Express)

// auth-middleware.ts
import type { Request, Response, NextFunction } from "express";

const VALID_KEYS = new Set(
  (process.env.MCP_API_KEYS ?? "").split(",").map((k) => k.trim()).filter(Boolean)
);

export function requireApiKey(req: Request, res: Response, next: NextFunction) {
  const authHeader = req.headers["authorization"];
  if (!authHeader?.startsWith("Bearer ")) {
    res.status(401).json({ error: "Missing Authorization header" });
    return;
  }
  const token = authHeader.slice(7);
  if (!VALID_KEYS.has(token)) {
    res.status(403).json({ error: "Invalid API key" });
    return;
  }
  next();
}

// Apply to the MCP endpoint:
// app.post("/mcp", requireApiKey, async (req, res) => { ... })

OAuth 2.1 with PKCE (Authorization Server Metadata)

// oauth-metadata.ts — expose the discovery document
import express from "express";
const app = express();

app.get("/.well-known/oauth-authorization-server", (_req, res) => {
  res.json({
    issuer: "https://mcp.example.com",
    authorization_endpoint: "https://mcp.example.com/oauth/authorize",
    token_endpoint: "https://mcp.example.com/oauth/token",
    registration_endpoint: "https://mcp.example.com/oauth/register",
    response_types_supported: ["code"],
    grant_types_supported: ["authorization_code", "refresh_token"],
    code_challenge_methods_supported: ["S256"],  // PKCE required
    token_endpoint_auth_methods_supported: ["none"],  // public clients
    scopes_supported: ["tools:read", "tools:write", "resources:read"],
  });
});

Deploying MCP Servers — Docker, Kubernetes, Railway

MCP servers are regular HTTP services (for remote transports) or CLI binaries (for stdio). Any container platform works. Below are production-ready configs for all three common deployment targets.

Docker

# Dockerfile
FROM node:22-alpine AS builder
WORKDIR /app
COPY package*.json ./
RUN npm ci --omit=dev
COPY . .
RUN npm run build

FROM node:22-alpine AS runtime
WORKDIR /app
ENV NODE_ENV=production
RUN addgroup -S mcp && adduser -S mcp -G mcp
COPY --from=builder --chown=mcp:mcp /app/dist ./dist
COPY --from=builder --chown=mcp:mcp /app/node_modules ./node_modules
USER mcp
EXPOSE 3000
HEALTHCHECK --interval=30s --timeout=5s --start-period=10s   CMD wget -qO- http://localhost:3000/health || exit 1
CMD ["node", "dist/http-server.js"]

Kubernetes

# mcp-server-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: mcp-postgres-server
  namespace: ai-tools
spec:
  replicas: 2
  selector:
    matchLabels:
      app: mcp-postgres-server
  template:
    metadata:
      labels:
        app: mcp-postgres-server
    spec:
      serviceAccountName: mcp-postgres-server
      securityContext:
        runAsNonRoot: true
        runAsUser: 1000
        seccompProfile:
          type: RuntimeDefault
      containers:
        - name: server
          image: registry.example.com/mcp-postgres-server:1.2.0
          ports:
            - containerPort: 3000
          env:
            - name: DATABASE_URL
              valueFrom:
                secretKeyRef:
                  name: mcp-postgres-secrets
                  key: database-url
            - name: MCP_API_KEYS
              valueFrom:
                secretKeyRef:
                  name: mcp-postgres-secrets
                  key: api-keys
          resources:
            requests:
              cpu: 100m
              memory: 128Mi
            limits:
              cpu: 500m
              memory: 256Mi
          readinessProbe:
            httpGet:
              path: /health
              port: 3000
            initialDelaySeconds: 5
            periodSeconds: 10
          livenessProbe:
            httpGet:
              path: /health
              port: 3000
            initialDelaySeconds: 15
            periodSeconds: 30
          securityContext:
            allowPrivilegeEscalation: false
            readOnlyRootFilesystem: true
            capabilities:
              drop: ["ALL"]
---
apiVersion: v1
kind: Service
metadata:
  name: mcp-postgres-server
  namespace: ai-tools
spec:
  selector:
    app: mcp-postgres-server
  ports:
    - port: 80
      targetPort: 3000

Railway (one-click deploy)

# railway.toml
[build]
builder = "dockerfile"
dockerfilePath = "Dockerfile"

[deploy]
startCommand = "node dist/http-server.js"
healthcheckPath = "/health"
healthcheckTimeout = 30
restartPolicyType = "on_failure"
restartPolicyMaxRetries = 3

[[services]]
name = "mcp-server"
port = 3000

[services.variables]
DATABASE_URL = "${DATABASE_URL}"
MCP_API_KEYS = "${MCP_API_KEYS}"

Production Patterns — Error Handling, Timeouts, Rate Limiting

Raw MCP server implementations need additional production hardening before they are safe to expose to real AI agents. Three areas require explicit attention: error propagation, request timeouts, and per-client rate limiting.

Structured Error Handling in TypeScript

// errors.ts — MCP-aware error wrapper
import { McpError, ErrorCode } from "@modelcontextprotocol/sdk/types.js";

export function toMcpError(err: unknown): McpError {
  if (err instanceof McpError) return err;

  if (err instanceof Error) {
    // Map well-known error types to MCP error codes
    if (err.message.includes("timeout") || err.message.includes("ETIMEDOUT")) {
      return new McpError(ErrorCode.InternalError, "Upstream service timed out");
    }
    if (err.message.includes("permission denied") || err.message.includes("EACCES")) {
      return new McpError(ErrorCode.InvalidRequest, "Permission denied");
    }
    // Scrub internal details — never expose stack traces to the LLM
    return new McpError(ErrorCode.InternalError, "Tool execution failed");
  }

  return new McpError(ErrorCode.InternalError, "Unknown error");
}

// Usage in a tool handler:
// try {
//   const result = await riskyOperation();
//   return { content: [{ type: "text", text: result }] };
// } catch (err) {
//   const mcpErr = toMcpError(err);
//   return { content: [{ type: "text", text: mcpErr.message }], isError: true };
// }

Per-Session Rate Limiting with Redis

// rate-limiter.ts — sliding window rate limiter using Redis
import { createClient } from "redis";

const redis = createClient({ url: process.env.REDIS_URL });
await redis.connect();

export async function checkRateLimit(
  sessionId: string,
  toolName: string,
  limitPerMinute = 60
): Promise<{ allowed: boolean; remaining: number; resetAt: number }> {
  const key = `mcp:rl:${sessionId}:${toolName}`;
  const now = Date.now();
  const windowStart = now - 60_000;

  const pipeline = redis.multi();
  pipeline.zRemRangeByScore(key, "-inf", windowStart.toString());
  pipeline.zAdd(key, { score: now, value: now.toString() });
  pipeline.zCard(key);
  pipeline.expire(key, 120);  // auto-expire after 2 minutes of inactivity

  const results = await pipeline.exec();
  const count = results[2] as number;
  const allowed = count <= limitPerMinute;
  const resetAt = now + 60_000;

  return { allowed, remaining: Math.max(0, limitPerMinute - count), resetAt };
}

Observability with OpenTelemetry

MCP servers in production need the same observability primitives as any other microservice: distributed traces, metrics, and structured logs. Because MCP tool calls originate from an AI agent, correlating tool execution time with the parent LLM request helps diagnose latency regressions and identify which tools are on the critical path.

// telemetry.ts — OpenTelemetry setup for an MCP server
import { NodeSDK } from "@opentelemetry/sdk-node";
import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-http";
import { OTLPMetricExporter } from "@opentelemetry/exporter-metrics-otlp-http";
import { PeriodicExportingMetricReader } from "@opentelemetry/sdk-metrics";
import { Resource } from "@opentelemetry/resources";
import { SEMRESATTRS_SERVICE_NAME, SEMRESATTRS_SERVICE_VERSION } from "@opentelemetry/semantic-conventions";

const sdk = new NodeSDK({
  resource: new Resource({
    [SEMRESATTRS_SERVICE_NAME]: "mcp-postgres-server",
    [SEMRESATTRS_SERVICE_VERSION]: process.env.npm_package_version ?? "0.0.0",
  }),
  traceExporter: new OTLPTraceExporter({
    url: process.env.OTEL_EXPORTER_OTLP_ENDPOINT ?? "http://localhost:4318/v1/traces",
  }),
  metricReader: new PeriodicExportingMetricReader({
    exporter: new OTLPMetricExporter({
      url: process.env.OTEL_EXPORTER_OTLP_ENDPOINT ?? "http://localhost:4318/v1/metrics",
    }),
    exportIntervalMillis: 30_000,
  }),
});

sdk.start();
process.on("SIGTERM", () => sdk.shutdown());

// --- Instrument tool calls ---
import { trace, SpanStatusCode } from "@opentelemetry/api";
const tracer = trace.getTracer("mcp-server");

export async function withToolSpan<T>(
  toolName: string,
  sessionId: string,
  fn: () => Promise<T>
): Promise<T> {
  return tracer.startActiveSpan(`mcp.tool.${toolName}`, async (span) => {
    span.setAttributes({
      "mcp.tool.name": toolName,
      "mcp.session.id": sessionId,
    });
    try {
      const result = await fn();
      span.setStatus({ code: SpanStatusCode.OK });
      return result;
    } catch (err) {
      span.setStatus({ code: SpanStatusCode.ERROR, message: String(err) });
      span.recordException(err as Error);
      throw err;
    } finally {
      span.end();
    }
  });
}

Security — Input Validation, Sandboxing, Prompt Injection Defense

MCP servers are a significant attack surface because they bridge the LLM's output (which can be adversarially manipulated) to real infrastructure (databases, file systems, external APIs). Three classes of attack require explicit mitigations: prompt injection via tool results, over-privileged tool scopes, and malicious inputs crafted by compromised upstream models.

// input-validation.ts — safe file path validation
import path from "path";
import { McpError, ErrorCode } from "@modelcontextprotocol/sdk/types.js";

const ALLOWED_ROOT = path.resolve(process.env.FILES_ROOT ?? "/data/files");

export function validateFilePath(userInput: string): string {
  // Resolve the path to eliminate any ../ traversal
  const resolved = path.resolve(ALLOWED_ROOT, userInput);

  // Ensure the resolved path is still within the allowed root
  if (!resolved.startsWith(ALLOWED_ROOT + path.sep) && resolved !== ALLOWED_ROOT) {
    throw new McpError(
      ErrorCode.InvalidParams,
      "Path traversal detected — access denied"
    );
  }

  return resolved;
}

// Sanitise untrusted content before returning to the LLM
export function sanitiseExternalContent(raw: string): string {
  return (
    raw
      // Remove instruction-injection patterns
      .replace(/[INST][sS]*?[/INST]/gi, "[content removed]")
      .replace(/<|system|>[sS]*?<|end|>/gi, "[content removed]")
      .replace(/ignore (all )?(previous|prior) instructions/gi, "[content removed]")
      // Strip HTML tags
      .replace(/<[^>]+>/g, "")
      // Truncate to reasonable size
      .slice(0, 10_000)
  );
}

Further Reading

• Model Context Protocol — Official Documentation — specification, transport reference, capability types, and the full protocol lifecycle
• modelcontextprotocol/typescript-sdk — GitHub — TypeScript SDK source, examples, and changelog for breaking transport API changes
• modelcontextprotocol/python-sdk — GitHub — Python SDK source, FastMCP high-level helpers, and asyncio patterns
• MCP Transport Reference — stdio, SSE, and HTTP Streamable transport comparison with migration guide
• OpenTelemetry Node.js Getting Started — auto-instrumentation, manual spans, and OTLP exporter configuration