Building a New MCP Server

Step-by-step guide to building, deploying, and registering a new MCP server for the Federal Frontier Platform.

Overview

Every MCP server in the platform follows the same pattern: a FastAPI application that exposes tools via JSON-RPC 2.0 at /jsonrpc, with health endpoints for Kubernetes probes. This guide walks through building a new server from scratch using the Kolla MCP Server as a reference implementation.

Project Structure

my-mcp-server/
  src/
    server.py          # FastAPI app, JSON-RPC handler, tool registry
    tools.py           # Tool definitions and handler functions
  tests/
    test_tools.py      # Unit tests with mocked backends
  Dockerfile
  requirements.txt
  pyproject.toml

Step 1: Define Your Tools

Each tool needs three things: a name, a JSON Schema for its input parameters, and a handler function that executes the tool.

# src/tools.py

TOOLS = [
    {
        "name": "myservice_list_items",
        "description": "List all items from MyService, optionally filtered by category.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "category": {
                    "type": "string",
                    "description": "Filter by category name"
                },
                "limit": {
                    "type": "integer",
                    "description": "Maximum number of results",
                    "default": 25
                }
            },
            "required": []
        }
    },
    {
        "name": "myservice_get_item",
        "description": "Get detailed information about a specific item.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "item_id": {
                    "type": "string",
                    "description": "The item identifier"
                }
            },
            "required": ["item_id"]
        }
    },
]

Step 2: Implement Handler Functions

Each tool name maps to a handler function. Handlers receive the tool arguments as a dictionary, perform the backend operation, and return a result.

# src/tools.py (continued)

import httpx

BACKEND_URL = os.environ.get("MYSERVICE_URL", "http://myservice:8080")

async def handle_list_items(arguments: dict) -> str:
    """List items from the backend service."""
    params = {}
    if "category" in arguments:
        params["category"] = arguments["category"]
    if "limit" in arguments:
        params["limit"] = arguments["limit"]

    async with httpx.AsyncClient() as client:
        resp = await client.get(f"{BACKEND_URL}/api/items", params=params)
        resp.raise_for_status()
        return json.dumps(resp.json(), indent=2)


async def handle_get_item(arguments: dict) -> str:
    """Get a specific item."""
    item_id = arguments["item_id"]
    async with httpx.AsyncClient() as client:
        resp = await client.get(f"{BACKEND_URL}/api/items/{item_id}")
        resp.raise_for_status()
        return json.dumps(resp.json(), indent=2)


# Map tool names to handlers
TOOL_HANDLERS = {
    "myservice_list_items": handle_list_items,
    "myservice_get_item": handle_get_item,
}

Step 3: Build the Server

The server is a FastAPI application with three endpoints: /jsonrpc for MCP tool calls, /health for liveness, and /ready for readiness.

# src/server.py

import json
import logging
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

from .tools import TOOLS, TOOL_HANDLERS

logger = logging.getLogger(__name__)
app = FastAPI(title="MyService MCP Server")


@app.get("/health")
async def health():
    return {"status": "ok"}


@app.get("/ready")
async def ready():
    # Check that the backend is reachable
    try:
        async with httpx.AsyncClient() as client:
            resp = await client.get(f"{BACKEND_URL}/health", timeout=5)
            resp.raise_for_status()
        return {"status": "ready"}
    except Exception as e:
        return JSONResponse(
            status_code=503,
            content={"status": "not ready", "error": str(e)}
        )


@app.post("/jsonrpc")
async def jsonrpc(request: Request):
    """Handle JSON-RPC 2.0 requests for MCP tools/list and tools/call."""
    body = await request.json()
    method = body.get("method")
    req_id = body.get("id")
    params = body.get("params", {})

    if method == "tools/list":
        return JSONResponse({
            "jsonrpc": "2.0",
            "id": req_id,
            "result": {"tools": TOOLS}
        })

    elif method == "tools/call":
        tool_name = params.get("name")
        arguments = params.get("arguments", {})

        handler = TOOL_HANDLERS.get(tool_name)
        if not handler:
            return JSONResponse({
                "jsonrpc": "2.0",
                "id": req_id,
                "error": {
                    "code": -32601,
                    "message": f"Unknown tool: {tool_name}"
                }
            })

        try:
            result_text = await handler(arguments)
            return JSONResponse({
                "jsonrpc": "2.0",
                "id": req_id,
                "result": {
                    "content": [{"type": "text", "text": result_text}]
                }
            })
        except Exception as e:
            logger.exception(f"Tool {tool_name} failed")
            return JSONResponse({
                "jsonrpc": "2.0",
                "id": req_id,
                "error": {
                    "code": -32000,
                    "message": str(e)
                }
            })

    else:
        return JSONResponse({
            "jsonrpc": "2.0",
            "id": req_id,
            "error": {
                "code": -32601,
                "message": f"Unknown method: {method}"
            }
        })

Step 4: Dockerfile

All MCP servers use python:3.11-slim as the base image:

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY src/ src/

EXPOSE 8080
CMD ["python", "-m", "uvicorn", "src.server:app", "--host", "0.0.0.0", "--port", "8080"]

requirements.txt:

fastapi>=0.104.0
uvicorn>=0.24.0
httpx>=0.25.0

Step 5: Unit Tests

Mock the backend and test handler logic:

# tests/test_tools.py

import pytest
import json
from unittest.mock import AsyncMock, patch
from httpx import Response

from src.tools import handle_list_items, handle_get_item


@pytest.mark.asyncio
async def test_list_items():
    mock_response = Response(200, json=[{"id": "1", "name": "item-1"}])
    with patch("src.tools.httpx.AsyncClient") as mock_client:
        mock_client.return_value.__aenter__ = AsyncMock(return_value=mock_client.return_value)
        mock_client.return_value.__aexit__ = AsyncMock(return_value=False)
        mock_client.return_value.get = AsyncMock(return_value=mock_response)

        result = await handle_list_items({"category": "compute"})
        parsed = json.loads(result)
        assert len(parsed) == 1
        assert parsed[0]["name"] == "item-1"

Step 6: Kubernetes Deployment

Create a deployment manifest in the GitOps repository:

# deploy/overlays/fmc/myservice-mcp/deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: myservice-mcp-server
  namespace: f3iai
  labels:
    app: myservice-mcp-server
spec:
  replicas: 1
  selector:
    matchLabels:
      app: myservice-mcp-server
  template:
    metadata:
      labels:
        app: myservice-mcp-server
    spec:
      containers:
        - name: server
          image: harbor.vitro.lan/ffp/myservice-mcp-server:v1.0.0
          ports:
            - containerPort: 8080
          env:
            - name: MYSERVICE_URL
              value: "http://myservice.f3iai.svc.cluster.local:8080"
          livenessProbe:
            httpGet:
              path: /health
              port: 8080
            initialDelaySeconds: 5
            periodSeconds: 10
          readinessProbe:
            httpGet:
              path: /ready
              port: 8080
            initialDelaySeconds: 5
            periodSeconds: 10
          resources:
            requests:
              memory: "128Mi"
              cpu: "100m"
            limits:
              memory: "256Mi"
              cpu: "500m"
---
apiVersion: v1
kind: Service
metadata:
  name: myservice-mcp-server
  namespace: f3iai
spec:
  selector:
    app: myservice-mcp-server
  ports:
    - port: 8080
      targetPort: 8080

Step 7: Register the Server

New servers must be registered in two places so Compass can discover their tools:

1. PostgreSQL `mcp_servers` Table

Insert a row into the mcp_servers table:

INSERT INTO mcp_servers (name, base_url, port, description, active)
VALUES (
    'myservice',
    'http://myservice-mcp-server.f3iai.svc.cluster.local',
    8080,
    'MyService MCP Server — items and categories',
    true
);

2. Compass API `MCP_SERVERS` Dict

Add the server to the MCP_SERVERS dictionary in the Compass API’s mcp_tools.py:

MCP_SERVERS = {
    # ... existing servers ...
    "myservice": {
        "name": "MyService",
        "base_url": "http://myservice-mcp-server.f3iai.svc.cluster.local:8080",
        "description": "Items and categories from MyService",
    },
}

After both registrations are complete, restart the Compass API pod. It will call tools/list on your server at startup and make the tools available to the LLM.

Checklist

Before deploying a new MCP server, verify: