Lesson 4 - Step 6 of 6

Building the Server

Build a complete MCP server with proper handshake handling, step by step.

Create: mcp_core.py

Now we'll build on the jsonrpc.py module from Lesson 3 to create mcp_core.py — the foundation for all MCP servers. Make sure jsonrpc.py is in the same directory.

Building on Lesson 3

This lesson imports and uses the JsonRpcRouter, error codes, and response helpers you built in Lesson 3. This is how real projects work — each module builds on previous ones.

How to Read Code Blocks

In this lesson:

  • Filename in header (e.g., mcp_core.py) = Write this code to that file
  • "Add after Section X" = Append this code to your existing file
  • bash blocks = Commands to run in your terminal
  • "Expected Output" in header = What you should see (don't write this)

Section 1: Imports and Connection State

Create a new file called mcp_core.py with these imports:

python mcp_core.py 
"""
mcp_core.py - MCP server with proper handshake handling.

This module builds on jsonrpc.py from Lesson 3.

Run: python mcp_core.py
Test: curl commands below
"""

import json
import sys
from http.server import HTTPServer, BaseHTTPRequestHandler
from enum import Enum
from typing import Any

# Import our JSON-RPC utilities from Lesson 3
from jsonrpc import (
    JsonRpcRouter,
    JsonRpcError,
    JsonRpcErrorCode,
    create_error_response,
)


# =============================================================================
# Connection State
# =============================================================================

class ConnectionState(Enum):
    """Track the state of an MCP connection."""
    DISCONNECTED = "disconnected"
    INITIALIZING = "initializing"
    READY = "ready"


# Server configuration
PROTOCOL_VERSION = "2024-11-05"
SERVER_NAME = "mcp-handshake-server"
SERVER_VERSION = "1.0.0"

# Connection state (global for simplicity)
state = ConnectionState.DISCONNECTED
client_info: dict[str, Any] = {}
client_capabilities: dict[str, Any] = {}

# Create our router using JsonRpcRouter from Lesson 3
router = JsonRpcRouter()

Section 2: Server Capabilities

Add this after Section 1:

python mcp_core.py - Add after Section 1 
# =============================================================================
# Server Capabilities
# =============================================================================

def get_server_capabilities() -> dict:
    """Return the capabilities this server supports."""
    return {
        "tools": {},  # Tools are implemented below
    }


def get_server_info() -> dict:
    """Return information about this server."""
    return {
        "name": SERVER_NAME,
        "version": SERVER_VERSION
    }

Section 3: MCP Handshake Handlers

Add the handshake handlers after Section 2. Notice how we use the @router.method() decorator from Lesson 3 to register handlers:

python mcp_core.py - Add after Section 2 
# =============================================================================
# MCP Handshake Handlers
# =============================================================================

@router.method("initialize")
def handle_initialize(params: dict) -> dict:
    """Handle the initialize request from a client."""
    global state, client_info, client_capabilities

    # Already initialized?
    if state != ConnectionState.DISCONNECTED:
        raise JsonRpcError(
            JsonRpcErrorCode.INVALID_REQUEST,
            "Already initialized"
        )

    # Store client information
    client_info = params.get("clientInfo", {})
    client_capabilities = params.get("capabilities", {})

    # Log the connection (stderr only for STDIO compatibility)
    name = client_info.get("name", "unknown")
    version = client_info.get("version", "unknown")
    print(f"[INIT] Client: {name} v{version}", file=sys.stderr)

    # Transition to INITIALIZING
    state = ConnectionState.INITIALIZING

    # Return our capabilities (router wraps this in a success response)
    return {
        "protocolVersion": PROTOCOL_VERSION,
        "capabilities": get_server_capabilities(),
        "serverInfo": get_server_info()
    }


@router.notification("notifications/initialized")
def handle_initialized(params: dict) -> None:
    """Handle the initialized notification from the client."""
    global state

    if state == ConnectionState.INITIALIZING:
        state = ConnectionState.READY
        print("[READY] Handshake complete!", file=sys.stderr)
    else:
        print(f"[WARN] Unexpected initialized notification in state: {state}", file=sys.stderr)
The Decorator Pattern

Using @router.method("initialize") automatically registers the handler. The router calls it when a matching request arrives, and wraps the return value in a proper JSON-RPC response. No manual response building needed!

Section 4: Placeholder Tool Handlers

Add these placeholder handlers after Section 3 (we'll expand them in Lesson 5):

python mcp_core.py - Add after Section 3 
# =============================================================================
# Tool Handlers (placeholder - expanded in Lesson 5)
# =============================================================================

@router.method("tools/list")
def handle_tools_list(params: dict) -> dict:
    """Return list of available tools."""
    return {
        "tools": [
            {
                "name": "echo",
                "description": "Echo a message back",
                "inputSchema": {
                    "type": "object",
                    "properties": {
                        "message": {"type": "string"}
                    },
                    "required": ["message"]
                }
            }
        ]
    }


@router.method("tools/call")
def handle_tools_call(params: dict) -> dict:
    """Execute a tool."""
    name = params.get("name", "")
    args = params.get("arguments", {})

    if name == "echo":
        return {
            "content": [{"type": "text", "text": args.get("message", "")}]
        }

    raise JsonRpcError(
        JsonRpcErrorCode.INVALID_PARAMS,
        f"Unknown tool: {name}"
    )

Section 5: State-Aware Request Handler

Add a wrapper that enforces the MCP state machine before delegating to the router:

python mcp_core.py - Add after Section 4 
# =============================================================================
# State-Aware Request Handler
# =============================================================================

def handle_request(request: dict) -> dict | None:
    """
    Process a JSON-RPC request with MCP state checking.

    The router handles method dispatch and response formatting.
    We just need to enforce the state machine.
    """
    method = request.get("method", "")
    request_id = request.get("id")

    # Initialize is always allowed
    if method == "initialize":
        return router.handle(request)

    # Initialized notification is always allowed
    if method == "notifications/initialized":
        return router.handle(request)

    # All other methods require READY state
    if state != ConnectionState.READY:
        if request_id is None:  # Notification
            print(f"[WARN] {method} received before ready", file=sys.stderr)
            return None
        return create_error_response(
            request_id=request_id,
            code=JsonRpcErrorCode.INVALID_REQUEST,
            message="Server not initialized"
        )

    # Delegate to the router for everything else
    return router.handle(request)

Section 6: HTTP Server

Add the HTTP server at the end:

python mcp_core.py - Add after Section 5 
# =============================================================================
# HTTP Server
# =============================================================================

class MCPHandler(BaseHTTPRequestHandler):
    """HTTP handler for MCP requests."""

    def log_message(self, format: str, *args) -> None:
        print(f"[HTTP] {args[0]}")

    def send_json(self, data: dict, status: int = 200) -> None:
        body = json.dumps(data, indent=2).encode()
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self) -> None:
        """Return server status."""
        self.send_json({
            "server": SERVER_NAME,
            "version": SERVER_VERSION,
            "state": state.value,
            "client": client_info or None
        })

    def do_POST(self) -> None:
        """Handle JSON-RPC requests."""
        length = int(self.headers.get("Content-Length", 0))
        body = self.rfile.read(length)

        try:
            request = json.loads(body)
        except json.JSONDecodeError as e:
            self.send_json(create_error_response(
                request_id=None,
                code=JsonRpcErrorCode.PARSE_ERROR,
                message=f"Parse error: {e}"
            ), 400)
            return

        if request.get("jsonrpc") != "2.0":
            self.send_json(create_error_response(
                request_id=request.get("id"),
                code=JsonRpcErrorCode.INVALID_REQUEST,
                message="jsonrpc must be '2.0'"
            ), 400)
            return

        response = handle_request(request)

        if response is None:
            self.send_response(204)
            self.end_headers()
        else:
            self.send_json(response)


def run_server(host: str = "localhost", port: int = 8080) -> None:
    """Start the MCP server."""
    server = HTTPServer((host, port), MCPHandler)
    print(f"MCP Server running on http://{host}:{port}")
    print("Press Ctrl+C to stop\n")
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\nShutting down...")


if __name__ == "__main__":
    run_server()

Try It

Make sure both jsonrpc.py (from Lesson 3) and mcp_core.py are in the same directory, then start your server:

bash 
python mcp_core.py

Test 1: Check Initial State

bash 
curl http://localhost:8080
json Expected Output 
{
  "server": "mcp-handshake-server",
  "version": "1.0.0",
  "state": "disconnected",
  "client": null
}

Test 2: Try tools/list Before Initialize (Should Fail)

bash 
curl -X POST http://localhost:8080 \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc":"2.0","id":1,"method":"tools/list"}'

Expected: Error code -32002 "Server not initialized"

Test 3: Send Initialize Request

bash 
curl -X POST http://localhost:8080 \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "id": 1,
    "method": "initialize",
    "params": {
      "protocolVersion": "2024-11-05",
      "capabilities": {},
      "clientInfo": {"name": "test-client", "version": "1.0"}
    }
  }'

Expected: Success response with server capabilities

Test 4: Send Initialized Notification

bash 
curl -X POST http://localhost:8080 \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc":"2.0","method":"notifications/initialized"}'

Expected: Empty response (HTTP 204), server logs "[READY] Handshake complete!"

Test 5: Now tools/list Works!

bash 
curl -X POST http://localhost:8080 \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc":"2.0","id":2,"method":"tools/list"}'

Expected: List of tools (including "echo")

What You've Built

Your MCP server now properly implements:

  • Module reuse — Imports JsonRpcRouter and helpers from Lesson 3
  • State tracking — DISCONNECTED → INITIALIZING → READY
  • Decorator-based handlers — Clean @router.method() syntax
  • Initialize request — Stores client info, returns capabilities
  • Initialized notification — Completes handshake, no response sent
  • Access control — Rejects requests before initialization

Your Project Structure

text 
your_project/
    jsonrpc.py      # Lesson 3: JSON-RPC utilities
    mcp_core.py     # Lesson 4: MCP handshake (imports jsonrpc)
Ready for More!

This mcp_core.py is the foundation for everything else. In the next lesson, you'll create a tools.py module that builds on this foundation to implement real, useful tools.

What's Next

In Lesson 5, you'll learn about Implementing Tools — the actions that MCP servers can perform. You'll create a tools.py module with a reusable tool registry pattern.