MCP client library

English | Русский

Overview

The MCP (Model Context Protocol) client library provides an asynchronous client for MCP servers. Connect to servers, call tools, list available tools, check server capabilities, and send ping requests.

Installation

Requires Python 3.13 or later.

pip install cloudru-qa-mcp-client

For structured logging with structlog, install the structlog extra:

pip install "cloudru-qa-mcp-client[structlog]"

Import the library as the module mcp_client.

Skills for AI agents

This library does not have its own skill yet. If your project uses other libraries from this suite with skills for AI agents, select and enable them from your project root (requires uv):

uvx library-skills

For Claude Code, add --claude.

Main components

BaseAsyncMCPClient

The main MCP client class supports:

• Connecting to MCP servers through connection handlers
• Calling server tools with arguments and metadata
• Listing available tools
• Checking server capabilities
• Sending ping requests to check the connection
• Configuring client options (timeouts, logging, retries, and more)

Connection handlers

The library supports the following connection handlers:

stdio_connection_handler

Create a standard input/output connection handler to start an MCP server.

streamable_http_connection_handler

Create an HTTP connection handler (Streamable HTTP transport) for a remote MCP server.

Example:

from mcp_client import BaseAsyncMCPClient
from mcp_client.connection_handlers import streamable_http_connection_handler

# Connect by URL
connection_handler = streamable_http_connection_handler("http://localhost:8000/mcp")

# Connect with a preconfigured HTTP client
from http_client import AsyncHttpClient
from http_client_retries import with_retries_failed_requests
from http_client_structlog import with_structlogger

http_client = AsyncHttpClient(
    "http://localhost:8000",
    with_retries_failed_requests(),
    with_structlogger(logger=logger),
)

connection_handler = streamable_http_connection_handler("http://localhost:8000/mcp", http_client)

client = BaseAsyncMCPClient(connection_handler)
async with client:
    tools = await client.list_tools()

Call results

ToolCallResponse

A wrapper for MCP tool call results. Provides properties to check success and access response content, structured data, and metadata.

Client options

Use these options to configure client behavior:

Timeouts

with_read_timeout

Set the read timeout for MCP client requests.

Example:

from mcp_client import BaseAsyncMCPClient
from mcp_client.options import with_read_timeout

client = BaseAsyncMCPClient(
    connection_handler,
    with_read_timeout(30.0),  # 30-second timeout
)

Logging

with_logger

Configure a logger for MCP client requests and responses.

Example:

import logging
from mcp_client import BaseAsyncMCPClient
from mcp_client.options import with_logger

logger = logging.getLogger(__name__)
client = BaseAsyncMCPClient(connection_handler, with_logger(logger))

with_structlogger

Configure structured logging for MCP client requests and responses. Requires structlog.

Example:

import structlog
from mcp_client import BaseAsyncMCPClient
from mcp_client.options import with_structlogger

logger = structlog.get_logger()
client = BaseAsyncMCPClient(connection_handler, with_structlogger(logger))

Metadata

with_default_metadata

Add default metadata to all client requests.

Example:

from mcp_client import BaseAsyncMCPClient
from mcp_client.options import with_default_metadata

client = BaseAsyncMCPClient(
    connection_handler, with_default_metadata({"client_version": "1.0.0", "environment": "production"})
)

Automatic identifiers

with_autogenerated_request_id

Add a new UUID to the x-request-id metadata of each call_tool, list_tools, and ping_server call.

with_autogenerated_traceparent

Add a new W3C traceparent to the metadata of each call_tool, list_tools, and ping_server call.

Explicit call metadata takes priority over defaults, and defaults take priority over generated values. Identifiers stay the same across retries of one call. One option can be reused across clients: each call receives its own values.

Connection initialization and internal SDK requests are excluded. get_server_capabilities reads cached server capabilities without sending a separate request.

For custom fields, a ClientOption can call client.add_metadata_provider(provider). The provider function returns a metadata dictionary and runs once before a public request is sent.

Retries

retry_if_tool_calling_fails

Configure retries for tool call errors.

Example:

from mcp_client import BaseAsyncMCPClient
from mcp_client.options import retry_if_tool_calling_fails

client = BaseAsyncMCPClient(
    connection_handler,
    retry_if_tool_calling_fails(3, 1.0, True),  # 3 attempts, 1-second interval, reraise the exception
)

retry_if_connection_fails

Configure retries for connection errors.

Example:

from mcp_client import BaseAsyncMCPClient
from mcp_client.options import retry_if_connection_fails

client = BaseAsyncMCPClient(
    connection_handler,
    retry_if_connection_fails(5, 2.0, False),  # 5 attempts, 2-second interval, do not reraise the exception
)

Usage

Example of using the MCP client library:

from mcp_client import BaseAsyncMCPClient
from mcp_client.connection_handlers import stdio_connection_handler

# Create a connection handler
connection_handler = stdio_connection_handler("path/to/mcp/server", ["--arg1", "value1"], {"ENV_VAR": "value"})

# Create a client with options
client = BaseAsyncMCPClient(
    connection_handler, with_read_timeout(30.0), with_default_metadata({"client_version": "1.0.0"})
)

# Use the client
async with client:
    # List tools
    tools = await client.list_tools()
    print(f"Available tools: {[tool.name for tool in tools.tools]}")

    # Call a tool
    response = await client.call_tool("echo", {"message": "Hello, MCP!"})
    if response.is_success:
        print(f"Tool response: {response.text_content()}")
    else:
        print(f"Tool call failed: {response.content}")

    # Check server capabilities
    capabilities = await client.get_server_capabilities()
    print(f"Server capabilities: {capabilities}")

    # Send a ping request
    ping_result = await client.ping_server()
    print(f"Ping result: {ping_result}")

Development

This library is developed in the qa-libraries monorepo alongside the other cloudru-qa-* libraries. You need uv, GNU Make and checkmake; run the commands from the repository root:

make sync                   # dependencies for all libraries from uv.lock
make test-mcp-client        # tests for this library
make typecheck-mcp-client   # mypy for this library
make check                  # formatting, Ruff, mypy, and tests for the whole suite

Tests run on Python 3.13 and 3.14. Each version uses a separate environment: make sync test-mcp-client PYTHON_VERSION=3.14.

Create a release with the Git tag mcp-client/v<version>. The tag version must match the version in packages/mcp-client/pyproject.toml.