Back to Workshop
MCP Challenge ~45 min

Build Your Own MCP Server

You've been adding tools to an existing server. Now build one from scratch. You'll learn how JSON-RPC 2.0 works under the hood, handle errors properly, and return structured content that agents can actually use.

What You'll Learn

JSON-RPC 2.0 Protocol

Understand every field in the MCP wire format

Error Handling

Return proper JSON-RPC errors that agents understand

Structured Content

Return both text and machine-readable data

Quick Refresher: What Makes an MCP Server?

An MCP server is just an HTTP server that speaks JSON-RPC 2.0. It has one endpoint (POST /message) and handles two methods: tools/list (what tools do you have?) and tools/call (run this tool). That's it.

tools/list

Agent calls this at startup to discover available tools.

Returns: tool names, descriptions, and input schemas (JSON Schema).

tools/call

Agent calls this when the AI decides to use a tool.

Returns: content (text/image), optional structuredContent, and isError flag.

The Challenge

Step 1

Create the Server Skeleton

Create a new directory for your MCP server. We'll build a "fun facts" server that serves random facts, quotes, and trivia. You'll implement the full JSON-RPC protocol from scratch.

terminal
# Create a new service directory
mkdir -p services/facts-server
cd services/facts-server

# Create the files we need
touch app.py requirements.txt Dockerfile

Start with requirements.txt:

services/facts-server/requirements.txt
fastapi==0.115.0
uvicorn==0.30.6
pydantic==2.9.0
Step 2

Implement the JSON-RPC Handler

This is the core of your MCP server. Every request comes in as a JSON-RPC message with a method field. You dispatch to the right handler based on that method. Let's break it down piece by piece.

First, the request/response models:

services/facts-server/app.py
import json
import random
import logging
from datetime import datetime
from typing import Any, Dict, Optional

import uvicorn
from fastapi import FastAPI
from pydantic import BaseModel

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI(title="Facts MCP Server", version="1.0.0")


# --- JSON-RPC 2.0 Models ---
# These match the JSON-RPC spec exactly. Every MCP server uses this format.

class JSONRPCRequest(BaseModel):
    jsonrpc: str = "2.0"       # Always "2.0"
    id: Optional[Any] = None    # Unique request identifier
    method: str                 # "tools/list" or "tools/call"
    params: Optional[Dict[str, Any]] = None


class JSONRPCError(BaseModel):
    code: int                   # Standard error codes: -32600, -32601, etc.
    message: str                # Human-readable error
    data: Optional[Any] = None  # Extra context


class JSONRPCResponse(BaseModel):
    jsonrpc: str = "2.0"
    id: Optional[Any] = None
    result: Optional[Any] = None   # Set on success
    error: Optional[JSONRPCError] = None  # Set on failure (never both!)

Key rule: A JSON-RPC response has either result or error, never both. The id in the response must match the id from the request.

Now the main handler that routes methods:

@app.post("/message")
async def handle_jsonrpc(request: JSONRPCRequest):
    """The single endpoint that handles all MCP communication."""
    try:
        # Validate protocol version
        if request.jsonrpc != "2.0":
            return JSONRPCResponse(
                id=request.id,
                error=JSONRPCError(
                    code=-32600,
                    message="Invalid Request",
                    data="Only JSON-RPC 2.0 is supported"
                )
            )

        # Route to the correct handler
        if request.method == "tools/list":
            return JSONRPCResponse(id=request.id, result=handle_tools_list())

        elif request.method == "tools/call":
            if not request.params or "name" not in request.params:
                return JSONRPCResponse(
                    id=request.id,
                    error=JSONRPCError(code=-32602, message="Missing 'name' in params")
                )
            result = await handle_tools_call(
                request.params["name"],
                request.params.get("arguments", {})
            )
            return JSONRPCResponse(id=request.id, result=result)

        else:
            return JSONRPCResponse(
                id=request.id,
                error=JSONRPCError(code=-32601, message=f"Unknown method: {request.method}")
            )

    except Exception as e:
        logger.error(f"Handler error: {e}")
        return JSONRPCResponse(
            id=request.id,
            error=JSONRPCError(code=-32603, message="Internal error", data=str(e))
        )
Step 3

Define Your Tool Manifest

The tool manifest is what agents read to understand what your server can do. Each tool needs a name, description, and an inputSchema (JSON Schema format). The description is especially important — it's how the AI decides when to use your tool.

services/facts-server/app.py (continued)
def handle_tools_list() -> Dict[str, Any]:
    """Return the manifest of all tools this server provides."""
    return {
        "tools": [
            {
                "name": "get_random_fact",
                "description": "Get a random interesting fact. Use this when the user asks for trivia, fun facts, or wants to learn something random.",
                "inputSchema": {
                    "type": "object",
                    "properties": {
                        "category": {
                            "type": "string",
                            "enum": ["science", "history", "nature", "space", "food"],
                            "description": "Category of fact to return"
                        }
                    },
                    "required": []
                }
            },
            {
                "name": "get_daily_quote",
                "description": "Get an inspirational or thought-provoking quote. Use when the user wants motivation or a quote of the day.",
                "inputSchema": {
                    "type": "object",
                    "properties": {},
                    "required": []
                }
            },
            {
                "name": "get_word_of_the_day",
                "description": "Get an interesting or unusual word with its definition. Great for vocabulary building.",
                "inputSchema": {
                    "type": "object",
                    "properties": {
                        "language": {
                            "type": "string",
                            "enum": ["en", "no"],
                            "default": "en",
                            "description": "Language for the word"
                        }
                    },
                    "required": []
                }
            }
        ]
    }

Pro tip: Write descriptions as if you're telling the AI when to use the tool. "Use this when the user asks for..." is much better than just "Returns a fact." The AI reads these descriptions to decide which tool to pick.

Step 4

Implement the Tool Logic

Now implement the actual tool functions. Each one returns a dict with content (for the AI to read), optional structuredContent (machine-readable data), and isError (flag for failures).

services/facts-server/app.py (continued)
# --- Fact Data ---
FACTS = {
    "science": [
        "Honey never spoils. Archaeologists found 3000-year-old honey in Egyptian tombs that was still edible.",
        "Octopuses have three hearts and blue blood.",
        "A teaspoon of a neutron star weighs about 6 billion tons.",
    ],
    "history": [
        "Cleopatra lived closer in time to the Moon landing than to the construction of the Great Pyramid.",
        "Oxford University is older than the Aztec Empire.",
    ],
    "nature": [
        "A group of flamingos is called a 'flamboyance'.",
        "Trees can communicate and share nutrients through underground fungal networks.",
    ],
    "space": [
        "A day on Venus is longer than its year.",
        "There are more stars in the universe than grains of sand on all of Earth's beaches.",
    ],
    "food": [
        "Bananas are berries, but strawberries are not.",
        "Peanuts are not actually nuts — they're legumes.",
    ],
}

QUOTES = [
    {"text": "The best way to predict the future is to invent it.", "author": "Alan Kay"},
    {"text": "Talk is cheap. Show me the code.", "author": "Linus Torvalds"},
    {"text": "Simplicity is the ultimate sophistication.", "author": "Leonardo da Vinci"},
    {"text": "First, solve the problem. Then, write the code.", "author": "John Johnson"},
]

WORDS = {
    "en": [
        {"word": "Petrichor", "definition": "The pleasant smell of rain on dry earth", "example": "After weeks of drought, the petrichor was intoxicating."},
        {"word": "Sonder", "definition": "The realization that each passerby has a life as vivid as your own", "example": "Walking through the busy station, she felt a deep sense of sonder."},
    ],
    "no": [
        {"word": "Koselig", "definition": "A feeling of coziness and warmth, like hygge but Norwegian", "example": "Vi hadde det veldig koselig foran peisen."},
        {"word": "Utepils", "definition": "A beer enjoyed outdoors in the sun", "example": "Endelig utepils-vær!"},
    ],
}


# --- Tool Call Router ---

async def handle_tools_call(tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
    """Route a tool call to the correct handler."""
    if tool_name == "get_random_fact":
        return get_random_fact(arguments)
    elif tool_name == "get_daily_quote":
        return get_daily_quote(arguments)
    elif tool_name == "get_word_of_the_day":
        return get_word_of_the_day(arguments)
    else:
        return {
            "content": [{"type": "text", "text": f"Unknown tool: {tool_name}"}],
            "isError": True
        }


def get_random_fact(arguments: dict) -> dict:
    category = arguments.get("category")
    if category and category not in FACTS:
        return {
            "content": [{"type": "text", "text": f"Unknown category '{category}'. Try: {list(FACTS.keys())}"}],
            "isError": True  # <-- Tell the agent this was an error!
        }

    if category:
        fact = random.choice(FACTS[category])
    else:
        all_facts = [f for facts in FACTS.values() for f in facts]
        fact = random.choice(all_facts)
        category = "random"

    return {
        "content": [{"type": "text", "text": fact}],
        "structuredContent": {  # <-- Bonus: machine-readable data
            "fact": fact,
            "category": category,
            "timestamp": datetime.now().isoformat()
        },
        "isError": False
    }


def get_daily_quote(arguments: dict) -> dict:
    quote = random.choice(QUOTES)
    return {
        "content": [{"type": "text", "text": f'"{quote["text"]}" — {quote["author"]}'}],
        "structuredContent": quote,
        "isError": False
    }


def get_word_of_the_day(arguments: dict) -> dict:
    language = arguments.get("language", "en")
    words = WORDS.get(language, WORDS["en"])
    word = random.choice(words)
    return {
        "content": [{
            "type": "text",
            "text": f"{word['word']}: {word['definition']}\nExample: {word['example']}"
        }],
        "structuredContent": word,
        "isError": False
    }
Step 5

Add a Health Check and Run It

Every good service needs a health endpoint. Add this and the server startup code:

services/facts-server/app.py (end of file)
@app.get("/health")
async def health_check():
    return {
        "status": "healthy",
        "service": "Facts MCP Server",
        "tools": 3,
        "timestamp": datetime.now().isoformat()
    }


if __name__ == "__main__":
    logger.info("Starting Facts MCP Server on port 8003...")
    uvicorn.run(app, host="0.0.0.0", port=8003)
Step 6

Dockerize and Wire It Up

Create a Dockerfile for your new server and add it to docker-compose.yml so it runs alongside everything else.

Dockerfile:

services/facts-server/Dockerfile
FROM python:3.11-slim

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

COPY . .

EXPOSE 8003
CMD ["python", "app.py"]

Add to docker-compose.yml:

docker-compose.yml
  # Add this alongside the other services
  facts-server:
    build:
      context: ./services/facts-server
      dockerfile: Dockerfile
    container_name: travel-weather-facts
    restart: unless-stopped
    networks:
      - travel-weather-network
    ports:
      - "8003:8003"
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8003/health"]
      interval: 30s
      timeout: 10s
      retries: 3
      start_period: 10s

Also update the MCP server environment to point to your new server:

docker-compose.yml — agent environment
  travel-agent:
    environment:
      - MCP_SERVER_URL=http://mcp-server:8000
      # If you did the multi-server challenge, add your facts server:
      - MCP_EXTRA_SERVERS=facts=http://facts-server:8003
Step 7

Test Your MCP Server

Build and test your brand new MCP server:

terminal
# Build and start everything
docker compose up -d --build

# Check health
curl http://localhost:8003/health

# List your tools
curl -X POST "http://localhost:8003/message" \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc": "2.0", "id": 1, "method": "tools/list"}'

# Get a random fact
curl -X POST "http://localhost:8003/message" \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0", "id": 2,
    "method": "tools/call",
    "params": {"name": "get_random_fact", "arguments": {"category": "space"}}
  }'

# Test error handling — try an unknown tool
curl -X POST "http://localhost:8003/message" \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0", "id": 3,
    "method": "tools/call",
    "params": {"name": "nonexistent_tool", "arguments": {}}
  }'

# Test an invalid category (should return isError: true)
curl -X POST "http://localhost:8003/message" \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0", "id": 4,
    "method": "tools/call",
    "params": {"name": "get_random_fact", "arguments": {"category": "bogus"}}
  }'
Stretch Goals

Take It Further

Now that you have a working MCP server, here are some ideas to extend it:

Connect to a real API

Replace the hardcoded facts with a real API. Try https://uselessfacts.jsph.pl/api/v2/facts/random for random facts or any other public API you like.

Add input validation

Validate arguments against the JSON Schema you defined in the tool manifest. Return clear error messages when inputs don't match.

Add caching with Redis

Check how the news server uses Redis caching in services/news-server/cache.py and add similar caching to avoid repeated API calls.

Compare with the news server

Explore services/news-server/app.py — it's a production-grade MCP server with RSS feeds, LLM summarization, Redis caching, and background prefetching. Great reference code.

You've Nailed It When...

tools/list returns your three tools with proper schemas
All three tools return both content and structuredContent
Invalid inputs return isError: true with helpful messages
The health check works and the server runs in Docker
An agent can discover and use your tools (if you did the multi-server challenge)