Lesson 10 - Step 1 of 6

Client Architecture

Understand how MCP clients manage connections and track requests.

Why Build a Client?

So far, you've built MCP servers. Now it's time to understand the other side: the MCP client. Building your own client gives you:

Deep understanding of the MCP protocol
A testing tool for your servers
The ability to integrate MCP into your own applications

Phase 4: Client Side

This lesson marks the beginning of Phase 4, where we flip perspectives from server to client. Understanding both sides makes you a complete MCP developer.

What You'll Build

In this lesson, you'll build a complete MCP client with an interactive REPL. By the end, you'll have:

stdio_client.py — A client that spawns and communicates with STDIO servers
mcp_repl.py — An interactive REPL for exploring MCP servers
main.py — A launcher script to run the REPL with any server

Each step builds incrementally, and you'll be able to test your client against any MCP server you've built in previous lessons.

Client Architecture Overview

An MCP client needs to manage several concerns:

diagram

┌─────────────────────────────────────────────────────────────┐
│                      MCP CLIENT                             │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ┌─────────────────┐     ┌─────────────────┐               │
│  │   Connection    │     │    Request      │               │
│  │   Manager       │     │    Tracker      │               │
│  │                 │     │                 │               │
│  │  - Start/stop   │     │  - Pending IDs  │               │
│  │  - Reconnect    │     │  - Callbacks    │               │
│  │  - Health check │     │  - Timeouts     │               │
│  └─────────────────┘     └─────────────────┘               │
│           │                      │                          │
│           └──────────┬───────────┘                          │
│                      │                                      │
│           ┌──────────▼──────────┐                          │
│           │    Transport Layer   │                          │
│           │  (STDIO or HTTP/SSE) │                          │
│           └──────────────────────┘                          │
│                      │                                      │
└──────────────────────│──────────────────────────────────────┘
                       │
                       ▼
               ┌───────────────┐
               │  MCP Server   │
               └───────────────┘

Connection Management

The connection manager handles the lifecycle of the server connection:

python

class MCPClient:
    """Base MCP client with connection management."""

    def __init__(self):
        self.connected = False
        self.initialized = False
        self.server_info = None
        self.server_capabilities = None

    def connect(self):
        """Establish connection to the server."""
        raise NotImplementedError

    def disconnect(self):
        """Close the connection."""
        raise NotImplementedError

    def is_connected(self) -> bool:
        """Check if connection is active."""
        return self.connected and self.initialized

Request Tracking

Since JSON-RPC allows responses to arrive out of order, we need to track pending requests by their ID:

python

import threading
from typing import Dict, Any, Optional, Callable

class RequestTracker:
    """Track pending requests and match responses."""

    def __init__(self):
        self._next_id = 1
        self._pending: Dict[int, dict] = {}
        self._lock = threading.Lock()

    def create_request(self, method: str, params: Optional[dict] = None) -> dict:
        """Create a new request with a unique ID."""
        with self._lock:
            request_id = self._next_id
            self._next_id += 1

            request: dict[str, Any] = {
                "jsonrpc": "2.0",
                "id": request_id,
                "method": method,
            }
            if params:
                request["params"] = params

            # Track this request
            self._pending[request_id] = {
                "request": request,
                "callback": None,
            }

            return request

    def resolve(self, response: dict) -> Optional[dict]:
        """Match a response to its request."""
        response_id = response.get("id")
        if response_id is None:
            return None

        with self._lock:
            pending = self._pending.pop(response_id, None)
            return pending

The Request-Response Cycle

Here's how a typical request flows through the client:

text

1. Application calls client.list_tools()
   │
   ▼
2. Client creates request: {"jsonrpc": "2.0", "id": 5, "method": "tools/list"}
   │
   ▼
3. Request ID 5 added to pending tracker
   │
   ▼
4. Request sent over transport (STDIO or HTTP)
   │
   ▼
5. Server processes request
   │
   ▼
6. Response arrives: {"jsonrpc": "2.0", "id": 5, "result": {...}}
   │
   ▼
7. Response matched to pending request by ID
   │
   ▼
8. Result returned to application

Thread Safety

If your client might be used from multiple threads, make sure to protect the request tracker with locks. The example above uses threading.Lock() for this purpose.

Client States

An MCP client moves through several states:

State	Description	Can Call Methods?
Disconnected	No connection to server	No
Connected	Transport connected, not initialized	Only `initialize`
Initializing	Handshake in progress	No
Ready	Fully initialized	Yes

Coming Up

In the next step, we'll implement a concrete STDIO client that can launch and communicate with MCP servers via subprocess.