Building an MCP Client from Scratch

An MCP client is the piece of software that connects a language model application to MCP servers, discovers the tools, resources, and prompts they expose, and routes calls between the model and those servers. Most engineers meet MCP through a ready-made client like an IDE extension or a chat app, but building your own mcp client from scratch is the fastest way to actually understand the protocol instead of treating it as a black box. This article walks through a working implementation in Python: connecting over stdio, listing capabilities, calling tools, wiring the results into an LLM conversation loop, and handling the failure modes you will hit in production.

What an MCP Client Actually Does

The Model Context Protocol splits responsibilities into three roles: a host application, an MCP client, and one or more MCP servers. The host is your app, say a CLI assistant or a backend service. The MCP client lives inside the host and owns exactly one connection to exactly one server, following the requests and messages over that connection according to the JSON-RPC 2.0 spec that MCP is built on. If your host talks to three servers (a filesystem server, a database server, a search server), it creates three separate client instances, one per connection.

A minimal mcp client needs to do five things:

• Open a transport (stdio for local processes, streamable HTTP for remote servers)
• Perform the initialize handshake and exchange capabilities
• List and call tools (tools/list, tools/call)
• List and read resources and prompts if the server offers them
• Handle notifications, errors, and cancellation cleanly

Everything else, model selection, conversation state, retries, is application logic layered on top. Keep that separation clear as you build: the client talks JSON-RPC to a server process, nothing more.

Setting Up Your Environment

You will need Python 3.10 or newer and the official mcp package, which ships the low-level client session, transport helpers, and type definitions for requests and results.

python3 -m venv .venv
source .venv/bin/activate
pip install mcp anthropic

The anthropic package is only needed for the chat loop example later in this article; the client itself only depends on mcp. If you want to test against a real server without writing one yourself, install a small reference server to point at:

pip install mcp-server-fetch

This gives you a server that fetches web pages, which is enough to exercise the full request cycle: connect, list tools, call a tool, read the result.

Connecting to an MCP Server Over stdio

Local MCP servers are just executables that read JSON-RPC messages from stdin and write responses to stdout. The client's job is to spawn that process, wire up the pipes, and speak the protocol over them. The mcp package's stdio_client handles process management; you supply the command and arguments.

import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def connect_and_list():
    server_params = StdioServerParameters(
        command="python3",
        args=["-m", "mcp_server_fetch"],
        env=None,
    )

    async with stdio_client(server_params) as (read_stream, write_stream):
        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()

            tools_result = await session.list_tools()
            for tool in tools_result.tools:
                print(f"{tool.name}: {tool.description}")

asyncio.run(connect_and_list())

Three things matter in that snippet. First, stdio_client is an async context manager: it starts the subprocess when entered and terminates it when the block exits, so you never leak a zombie process. Second, session.initialize() is not optional. It performs the handshake where client and server exchange protocol versions and capability flags; skipping it means the server may reject every subsequent call. Third, list_tools() returns structured Tool objects with name, description, and an inputSchema (JSON Schema) you will need for validating arguments before you call the tool.

Run the script and you should see the fetch server's single tool printed with its description. If nothing prints, check that the server command is on your PATH and that you are not swallowing stderr, since server startup errors show up there, not in the JSON-RPC stream.

Discovering and Calling Tools

Tool discovery and invocation is the core loop of any mcp client. Once you have the tool list, calling one is a single request with the tool name and an arguments object that must satisfy the tool's inputSchema.

async def call_fetch_tool(session: ClientSession, url: str) -> str:
    result = await session.call_tool(
        name="fetch",
        arguments={"url": url, "max_length": 5000},
    )

    if result.isError:
        raise RuntimeError(f"Tool call failed: {result.content}")

    text_chunks = [
        block.text for block in result.content if block.type == "text"
    ]
    return "\n".join(text_chunks)

A few details that trip people up when they build their first client:

• result.content is a list of content blocks, not a single string. A tool can return text, images, and embedded resources in one response, so always filter by block.type before reading .text.
• result.isError is a boolean on the result, not an exception. MCP treats tool execution failures (a bad URL, a timeout inside the tool) as normal results with isError=True, reserving JSON-RPC-level exceptions for protocol errors like an unknown tool name. Your client code has to check both layers.
• Arguments are validated against inputSchema on the server side, but validating them client-side first (with a library like jsonschema) saves a network round trip and gives you a better error message to show the user or the model.

If you are feeding tool definitions to an LLM for function calling, convert the MCP inputSchema directly into the provider's tool format. Anthropic's Messages API, for example, accepts JSON Schema almost as-is under the input_schema field of a tool definition, so the mapping is close to a pass-through.

Handling Resources and Prompts

Tools are the most visible MCP primitive, but a complete mcp client also needs to handle resources (read-only data the server exposes, like files or query results) and prompts (server-defined templates for common tasks).

async def read_resources(session: ClientSession):
    resources = await session.list_resources()
    for resource in resources.resources:
        print(resource.uri, resource.mimeType)

    if resources.resources:
        first = resources.resources[0]
        content = await session.read_resource(first.uri)
        for block in content.contents:
            if hasattr(block, "text"):
                print(block.text[:200])

async def use_prompt(session: ClientSession, prompt_name: str):
    prompts = await session.list_prompts()
    names = [p.name for p in prompts.prompts]
    if prompt_name not in names:
        return None

    result = await session.get_prompt(prompt_name, arguments={})
    return result.messages

Resources are addressed by URI, and the scheme is server-defined: a filesystem server might use file://, a database server might invent its own scheme like postgres://schema/table. Your client should not assume a scheme; treat the URI as opaque and pass it straight back to read_resource.

Prompts return a list of messages already shaped for a chat completion, meaning the server has decided the roles and content for you. This is useful for exposing curated workflows (a server-side "summarize this ticket" prompt) without duplicating that logic in every client that connects.

Not every server implements resources or prompts. Check session.server_info.capabilities (or catch the resulting error) before calling list_resources or list_prompts on a server that never advertised support for them.

Wiring the Client into a Chat Loop

An mcp client is only useful once it is feeding tool results back into a model. Here is a minimal loop using the Claude Messages API: send the user's message plus the MCP tool definitions, execute any tool the model asks for through the session, and hand the result back.

import json
from anthropic import Anthropic

def mcp_tools_to_anthropic_format(tools_result):
    return [
        {
            "name": t.name,
            "description": t.description or "",
            "input_schema": t.inputSchema,
        }
        for t in tools_result.tools
    ]

async def chat_with_tools(session: ClientSession, user_message: str):
    client = Anthropic()
    tools_result = await session.list_tools()
    anthropic_tools = mcp_tools_to_anthropic_format(tools_result)

    messages = [{"role": "user", "content": user_message}]

    while True:
        response = client.messages.create(
            model="claude-sonnet-4-5",
            max_tokens=1024,
            tools=anthropic_tools,
            messages=messages,
        )

        if response.stop_reason != "tool_use":
            final_text = "".join(
                block.text for block in response.content if block.type == "text"
            )
            return final_text

        messages.append({"role": "assistant", "content": response.content})
        tool_results = []

        for block in response.content:
            if block.type != "tool_use":
                continue
            call_result = await session.call_tool(
                name=block.name, arguments=block.input
            )
            output_text = "\n".join(
                c.text for c in call_result.content if c.type == "text"
            )
            tool_results.append(
                {
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": output_text,
                    "is_error": call_result.isError,
                }
            )

        messages.append({"role": "user", "content": tool_results})

The loop is deliberately plain: request the model's response, check whether it wants a tool, run the tool through the MCP session, append the result, and go around again. This is the same shape used by every agent framework that supports MCP; the framework code just adds retries, streaming, and multi-server routing around this core.

Error Handling and Timeouts

Real servers hang, crash, and return malformed data. A client that assumes the happy path will wedge your whole application the first time a tool call blocks on a slow network resource. Wrap calls with asyncio.wait_for and catch both transport-level and protocol-level failures separately.

from mcp.shared.exceptions import McpError

async def safe_call_tool(session: ClientSession, name: str, arguments: dict, timeout: float = 30.0):
    try:
        return await asyncio.wait_for(
            session.call_tool(name=name, arguments=arguments),
            timeout=timeout,
        )
    except asyncio.TimeoutError:
        raise RuntimeError(f"Tool '{name}' timed out after {timeout}s")
    except McpError as e:
        raise RuntimeError(f"Protocol error calling '{name}': {e}")

Also plan for the server process dying mid-session. If you spawned it with stdio_client, the context manager will surface a broken pipe on the next read; catch that at the call site and decide whether to reconnect, fall back to a cached tool list, or surface the failure to the user. Do not silently retry a tool call that mutates state (a write, a send-email call) without idempotency guarantees from the server; retrying reads is almost always safe, retrying writes usually is not.

Connecting Over HTTP for Remote Servers

Everything above uses stdio, which only works for servers running as a local subprocess. Remote MCP servers use the Streamable HTTP transport instead, and the client-side API is nearly identical, you just swap the transport.

from mcp.client.streamable_http import streamablehttp_client

async def connect_remote(url: str, headers: dict):
    async with streamablehttp_client(url, headers=headers) as (read_stream, write_stream, _):
        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()
            tools = await session.list_tools()
            return tools

For authenticated remote servers, MCP defines an OAuth-based authorization flow. Pass the bearer token in the headers dict once you have completed that flow separately; the session code above does not change. The practical difference from stdio is operational, not architectural: you now need to handle reconnects across network blips, respect server-side rate limits, and treat the connection as something that can be shared across requests rather than owned by a single process lifetime.

Testing Your MCP Client

Test the client against a minimal server you control before pointing it at anything third-party. Write a tiny server with one tool that returns a fixed string, then assert your client's call_tool wrapper returns exactly what you expect, including the error path.

import pytest

@pytest.mark.asyncio
async def test_call_tool_returns_text(mcp_session):
    result = await mcp_session.call_tool(name="echo", arguments={"text": "hello"})
    assert not result.isError
    assert result.content[0].text == "hello"

@pytest.mark.asyncio
async def test_call_tool_timeout(mcp_session_slow):
    with pytest.raises(RuntimeError, match="timed out"):
        await safe_call_tool(mcp_session_slow, "slow_echo", {"text": "hi"}, timeout=0.1)

A mcp_session fixture that spins up an in-process test server over stdio (or an anyio memory stream, which the mcp package supports directly) keeps these tests fast and free of network flakiness. Add one test per failure mode you handled in the previous section: timeout, malformed arguments, server crash, and a tool result that omits the content field entirely, since not every server implementation is equally strict about the spec.

Common Pitfalls

• Forgetting to call `initialize()`. Every session must complete the handshake before any other request; servers are within spec to reject calls made before it.
• Assuming one client handles many servers. The ClientSession object is bound to a single transport. If your host needs to talk to five servers, instantiate five sessions and keep a registry keyed by server name.
• Not re-fetching tool lists. Servers can send a tools/list_changed notification when their tool set changes at runtime (a plugin loads, a feature flag flips). A client that caches the tool list once at startup will silently miss new tools until restarted.
• Blocking the event loop. The whole client API is async; running it inside a synchronous framework without asyncio.run or a proper event loop bridge causes deadlocks, especially with stdio transports where reads and writes must interleave.
• Treating `isError` and exceptions the same way. Protocol errors raise; tool execution errors do not. Mixing up the two means you either crash on a normal tool failure or silently ignore a broken connection.

FAQ

What language should I use to build an MCP client? Python and TypeScript have official SDKs maintained alongside the spec and are the most common choices. Both expose the same core primitives shown here: a session object, transport helpers, and typed request/result classes. Other languages have community SDKs of varying completeness, so check current coverage before committing to one for a production system.

Do I need to build a client if I already use Claude Desktop or an IDE with MCP support? No. Those applications embed their own mcp client already. Build your own only when you are writing a custom host application, an agent framework, a backend service, or anything that needs programmatic control over tool discovery and invocation beyond what a pre-built client exposes.

How is an MCP client different from an MCP server? The server exposes tools, resources, and prompts; the client consumes them on behalf of a host application. They speak the same JSON-RPC message format but take opposite roles in every exchange: the client sends requests like tools/call, the server sends results and notifications.

Can one client connect to multiple servers at once? A single ClientSession maps to a single transport and a single server. To talk to multiple servers, your host application creates one session per server and aggregates their tool lists before presenting them to the model, typically prefixing tool names by server to avoid collisions.

What happens if a tool call takes too long? Nothing built into the protocol enforces a timeout, so it is entirely the client's responsibility. Wrap calls in asyncio.wait_for (or your language's equivalent) and decide on a sane default, generally somewhere between 10 and 60 seconds depending on the tool, with longer-running tools using progress notifications instead of a single blocking call.

Is stdio or HTTP better for a new server? Use stdio for local tools that run alongside the host process, like filesystem or shell access, since there is no network layer to secure. Use Streamable HTTP for anything shared across users or deployed as a service, since it supports authentication and can run independently of the client's lifecycle.