Building an MCP Client in Python

To use the Model Context Protocol from Python, install the official SDK with pip install "mcp[cli]", create a Client with either a URL (Streamable HTTP) or a StdioServerParameters (a local server it launches), and open it with async with. Inside that block, await client.list_tools() returns each tool’s name, description and input schema, and await client.call_tool(name, arguments) returns a result with content for a model to read, structured_content for your code, and is_error. To put a model in charge, hand it the tool list, run the tools it asks for through the client, and pass the results back until it stops asking. The rest of this guide builds that, step by step, with code checked against version 2 of the SDK.

This is the client side. Writing the server is covered in how to build an MCP server; what tools, resources and prompts are is in the three building blocks; and if you would rather let OpenAI’s Agents SDK be the client, see MCP with OpenAI.

Install the SDK, and know which version you have

The MCP Python SDK’s README↗ (opens in a new tab) says version 2 is the current stable line, a major rework that supports the 2026-07-28 revision of the protocol and every earlier one. It needs Python 3.10 or newer. Version 1 lives on a v1.x branch and still gets security fixes; because a plain pip install mcp now installs 2.x, the README suggests pinning mcp>=1.28,<2 until you migrate.

uv add "mcp[cli]" anthropic     # or: pip install "mcp[cli]" anthropic
python -c "import importlib.metadata as m; print(m.version('mcp'))"

Check that number before you copy code from anywhere. Most tutorials written before version 2 open a ClientSession over stdio_client, call initialize() themselves and read inputSchema in camel case. In version 2 one Client object does all of that, the fields are snake case, and the session is still there as client.session if you need the low-level layer.

A server to talk to

You need something on the other end. This one keeps a two-line task list, and raising ToolError is how the SDK turns a refusal into a result the model can read. Save it as server.py.

from mcp.server import MCPServer
from mcp.server.mcpserver.exceptions import ToolError

mcp = MCPServer("Tasks")
TASKS = {"T-1": "Write release notes"}


@mcp.tool()
def list_tasks() -> dict[str, str]:
    """List open tasks by id."""
    return TASKS


@mcp.tool()
def complete_task(task_id: str) -> str:
    """Mark a task done."""
    if task_id not in TASKS:
        raise ToolError(f"No task {task_id}. Call list_tasks to see the ids.")
    del TASKS[task_id]
    return f"Completed {task_id}"


if __name__ == "__main__":
    mcp.run()

Connect over stdio, list tools, call one

Over stdio, your client starts the server as a child process and speaks to it through its standard input and output. Describe the process with StdioServerParameters; entering the async with block starts it and leaving the block shuts it down.

import sys

import anyio

from mcp import Client, StdioServerParameters
from mcp.types import TextContent

server = StdioServerParameters(command=sys.executable, args=["server.py"])


async def main() -> None:
    async with Client(server) as client:
        print(client.server_info.name, client.protocol_version)

        listed = await client.list_tools()
        for tool in listed.tools:
            print(tool.name, "-", tool.description, tool.input_schema)

        result = await client.call_tool("complete_task", {"task_id": "T-9"})
        if result.is_error:
            for block in result.content:
                if isinstance(block, TextContent):
                    print("Refused:", block.text)
        else:
            print(result.structured_content)


if __name__ == "__main__":
    anyio.run(main)

Run python client.py and it prints the server’s name, the protocol version the two sides agreed, both tools with their schemas, and then “Refused: … No task T-9. Call list_tasks to see the ids.” Using sys.executable makes the child use the same Python, and so the same installed packages, as the client.

One surprise on this transport: the child does not inherit your environment. The SDK’s client transports guide↗ (opens in a new tab) says it gets a short allow-list, such as PATH and HOME, so that nothing sensitive leaks into a process you may not have written. A server that needs an API key gets it only if you pass it with env={...}.

Connect over Streamable HTTP

For a remote server, pass the URL instead. Start the same server over HTTP with uv run mcp run server.py --transport streamable-http and it listens on http://localhost:8000/mcp. Everything inside the block stays the same.

import os

import anyio
import httpx2

from mcp import Client
from mcp.client.streamable_http import streamable_http_client


async def main() -> None:
    # The simple case: a URL means Streamable HTTP
    async with Client("http://localhost:8000/mcp") as client:
        print([t.name for t in (await client.list_tools()).tools])

    # With a bearer token: headers live on your own HTTP client
    headers = {"Authorization": f"Bearer {os.environ['MCP_TOKEN']}"}
    async with httpx2.AsyncClient(headers=headers, timeout=httpx2.Timeout(30.0, read=300.0)) as http:
        async with Client(streamable_http_client("https://example.com/mcp", http_client=http)) as client:
            print((await client.call_tool("list_tasks", {})).structured_content)


if __name__ == "__main__":
    anyio.run(main)

Two details catch people moving from version 1. streamable_http_client no longer takes headers= or timeout=; anything HTTP-shaped goes on an httpx2.AsyncClient you create, enter and close yourself. And the transport follows a redirect only within the same origin, or from http to https on the same host; anything else fails with a message naming the URL to use instead.

For servers that sign people in with OAuth, the SDK has an OAuthClientProvider that plugs into the same httpx2.AsyncClient as its auth. The OAuth clients page↗ (opens in a new tab) walks through its four inputs: the server URL, your client’s registration details, somewhere to store tokens, and two handlers for the moment a person opens the browser and comes back. What that sign-in looks like from the person’s side is in how MCP sign-in works.

Read the result properly

• content is a list of blocks, and a block may be text, an image, audio, a resource link or an embedded resource. Check the type with isinstance before you read .text. This is what a model should see.
• structured_content is the tool’s return value as JSON, matching the tool’s output schema when it has one. This is what your code should read, instead of parsing text.
• is_error is the one to check first. A tool that raises does not raise in your client: it comes back as a normal result with is_error=True and the message in content. Even an unknown tool name comes back this way.
• An exception, MCPError, is raised only when the server answers with a JSON-RPC error instead of a result. Treat that as a broken connection or a protocol problem, not as something to show the model.
• List calls may be paged. Every list_* method takes a cursor= argument and returns next_cursor; loop until it is None if a server has many tools.

Which protocol version you get

You never pick a version by hand. By default, as the SDK’s protocol versions page↗ (opens in a new tab) explains, entering the block sends one server/discover probe at the newest version; a 2026-07-28 server answers it, and an older server returns an error, so the client falls back to the classic initialize handshake. client.protocol_version tells you which you got. Pass mode="legacy" when you need the server to call you back mid-tool, for sampling or a form-style elicitation, because that channel only exists on handshake-era connections. What changed between revisions, and why, is in MCP specification changes.

Put a model in the loop

A client on its own calls the tools you name. An agent lets a model choose. The shape is the same for any provider: give the model the tool list, run each tool call it makes through the MCP client, return the results, and repeat until it answers in plain text. This example uses Claude through the Anthropic SDK, following Anthropic’s tool use documentation↗ (opens in a new tab); the MCP half would not change with another model.

import sys

import anthropic
import anyio

from mcp import Client, StdioServerParameters
from mcp.types import TextContent

server = StdioServerParameters(command=sys.executable, args=["server.py"])
claude = anthropic.AsyncAnthropic()  # reads ANTHROPIC_API_KEY


def as_text(result) -> str:
    return "\n".join(b.text for b in result.content if isinstance(b, TextContent))


async def run(prompt: str) -> str:
    async with Client(server) as mcp:
        listed = await mcp.list_tools()
        tools = [
            {"name": t.name, "description": t.description or "", "input_schema": t.input_schema}
            for t in listed.tools
        ]
        messages = [{"role": "user", "content": prompt}]

        for _ in range(10):  # a hard stop, whatever the model does
            response = await claude.messages.create(
                model="claude-opus-5",
                max_tokens=16000,
                tools=tools,
                messages=messages,
            )
            if response.stop_reason != "tool_use":
                return "".join(b.text for b in response.content if b.type == "text")

            messages.append({"role": "assistant", "content": response.content})
            results = []
            for block in response.content:
                if block.type == "tool_use":
                    result = await mcp.call_tool(block.name, block.input)
                    results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": as_text(result),
                        "is_error": result.is_error,
                    })
            messages.append({"role": "user", "content": results})

        return "Stopped after 10 turns."


if __name__ == "__main__":
    print(anyio.run(run, "Complete every open task, then tell me what you did."))

Four choices in that loop are deliberate. The MCP tool definition maps field for field onto the model’s tool format, so there is no hand-written schema to drift. Every tool call in one response is answered in one message, which keeps parallel calls working. The is_error flag goes back with the text, so the model reads “No task T-9” as a refusal and can recover by listing tasks first. And the loop has a ceiling, because a model that keeps calling tools will otherwise keep spending.

Before this touches anything real, add an approval step for tools that change data: check block.name against a list of writes and ask the person before call_tool runs. The reasons, and the other habits worth keeping, are in MCP security best practices.

Pointing it at a task board

The same client works against any remote server. fenbs, a task board where AI assistants are members with roles, serves its MCP tools at https://fenbs.ai/api/mcp. For a script, issue a token by hand under Settings, “Connect an AI assistant”, tick only the scopes the job needs, and send it as the bearer header in the HTTP example above. Call fenbs_whoami first: it says which boards the token can reach and the role it holds on each. A refusal comes back as an is_error result that names the permission that was missing, which is exactly what the loop above passes back to the model, and every change appears in the board’s History under the assistant’s name.