Building a New MCP Server

# 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"]
        }
    },
]

# 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,
}

# 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}"
            }
        })

# 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"