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.
Understand every field in the MCP wire format
Return proper JSON-RPC errors that agents understand
Return both text and machine-readable data
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.
Agent calls this at startup to discover available tools.
Returns: tool names, descriptions, and input schemas (JSON Schema).
Agent calls this when the AI decides to use a tool.
Returns: content (text/image), optional structuredContent, and isError flag.
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.
# 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:
fastapi==0.115.0 uvicorn==0.30.6 pydantic==2.9.0
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:
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)) )
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.
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.
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).
# --- 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 }
Every good service needs a health endpoint. Add this and the server startup code:
@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)
Create a Dockerfile for your new server and add it to docker-compose.yml so it runs alongside everything else.
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:
# 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:
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 Build and test your brand new MCP server:
# 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"}} }'
Now that you have a working MCP server, here are some ideas to extend it:
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.
Validate arguments against the JSON Schema you defined in the tool manifest. Return clear error messages when inputs don't match.
Check how the news server uses Redis caching in services/news-server/cache.py and add similar caching to avoid repeated API calls.
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.
tools/list returns your three tools with proper schemas content and structuredContent isError: true with helpful messages