Convert REST API to MCP Server: Production-Grade TypeScript & Python Migration Guide

// src/server.ts - Production MCP Server (TypeScript)
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";
import axios from "axios";

const restClient = axios.create({
  baseURL: process.env.REST_API_BASE_URL,
  timeout: 30000,
  headers: { Authorization: `Bearer ${process.env.REST_API_TOKEN}` }
});

const server = new Server(
  { name: "crm-mcp-server", version: "1.0.0" },
  { capabilities: { tools: {} } }
);

server.setRequestHandler(ListToolsRequestSchema, async () => ({
  tools: [
    {
      name: "get_contact",
      description: "Retrieve a single CRM contact by email with enrichment context",
      inputSchema: {
        type: "object",
        properties: { email: { type: "string", format: "email" } },
        required: ["email"],
        additionalProperties: false
      }
    },
    {
      name: "list_contacts",
      description: "List CRM contacts with optional status filter and pagination",
      inputSchema: {
        type: "object",
        properties: {
          status: { type: "string", enum: ["active", "churned"] },
          limit: { type: "integer", minimum: 1, maximum: 100, default: 50 },
          cursor: { type: "string" }
        },
        additionalProperties: false
      }
    }
  ]
}));

server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;
  try {
    if (name === "get_contact") {
      const { data } = await restClient.get(`/contacts/${args.email}`);
      return { content: [{ type: "text", text: JSON.stringify(data) }] };
    }
    if (name === "list_contacts") {
      const params = new URLSearchParams();
      if (args.status) params.set("status", args.status);
      params.set("limit", String(args.limit ?? 50));
      if (args.cursor) params.set("cursor", args.cursor);
      const { data } = await restClient.get(`/contacts?${params}`);
      return { content: [{ type: "text", text: JSON.stringify(data) }] };
    }
    throw new Error(`Unknown tool: ${name}`);
  } catch (error) {
    return {
      content: [{ type: "text", text: `Error: ${error.message}` }],
      isError: true
    };
  }
});

async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
}
main().catch(console.error);

# server.py - Production MCP Server (Python)
import os
import httpx
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
from pydantic import BaseModel, Field

server = Server("crm-mcp-server")
client = httpx.AsyncClient(
    base_url=os.environ["REST_API_BASE_URL"],
    timeout=30.0,
    headers={"Authorization": f"Bearer {os.environ['REST_API_TOKEN']}"}
)

class GetContactInput(BaseModel):
    email: str = Field(..., description="Contact email address")

class ListContactsInput(BaseModel):
    status: str | None = Field(None, enum=["active", "churned"])
    limit: int = Field(50, ge=1, le=100)
    cursor: str | None = None

@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="get_contact",
            description="Retrieve a single CRM contact by email",
            inputSchema=GetContactInput.model_json_schema()
        ),
        Tool(
            name="list_contacts",
            description="List CRM contacts with status filter and pagination",
            inputSchema=ListContactsInput.model_json_schema()
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    try:
        if name == "get_contact":
            args = GetContactInput(**arguments)
            resp = await client.get(f"/contacts/{args.email}")
            return [TextContent(type="text", text=resp.text)]
        if name == "list_contacts":
            args = ListContactsInput(**arguments)
            params = {"limit": args.limit}
            if args.status: params["status"] = args.status
            if args.cursor: params["cursor"] = args.cursor
            resp = await client.get("/contacts", params=params)
            return [TextContent(type="text", text=resp.text)]
        raise ValueError(f"Unknown tool: {name}")
    except Exception as e:
        return [TextContent(type="text", text=f"Error: {str(e)}")]

async def main():
    async with stdio_server() as (read, write):
        await server.run(read, write, server.create_initialization_options())

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

# Dockerfile - Production MCP Server
FROM node:20-alpine AS builder
WORKDIR /app
COPY package*.json ./
RUN npm ci --only=production
COPY . .
RUN npm run build

FROM node:20-alpine
WORKDIR /app
COPY --from=builder /app/dist ./dist
COPY --from=builder /app/node_modules ./node_modules
ENV NODE_ENV=production
ENV REST_API_BASE_URL=http://rest-api:8080
EXPOSE 3000
HEALTHCHECK --interval=30s --timeout=3s --retries=3 \
  CMD node -e "process.exit(0)"
CMD ["node", "dist/server.js"]

# docker-compose.yml - MCP Server + REST API (Zero-Downtime)
version: '3.8'
services:
  rest-api:
    image: mycompany/crm-rest-api:latest
    ports: ["8080:8080"]
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8080/health"]
      interval: 10s
      timeout: 3s
      retries: 3
  mcp-server:
    build: .
    depends_on:
      rest-api:
        condition: service_healthy
    environment:
      - REST_API_BASE_URL=http://rest-api:8080
      - REST_API_TOKEN=${REST_API_TOKEN}
    restart: unless-stopped
    deploy:
      resources:
        limits:
          memory: 256M
          cpus: '0.5'