Клиентские транспорты

Каждый Client общается со своим сервером через транспорт — то, что на самом деле переносит сообщения.

Настраивать его отдельно не нужно. Client принимает один позиционный аргумент и определяет транспорт по его типу.

Серверная сторона каждого из них (то, что делает mcp.run() и что вы развёртываете) описана на странице Запуск сервера.

В памяти

Передайте сам объект сервера:

from mcp import Client
from mcp.server import MCPServer

mcp = MCPServer("Bookshop")


@mcp.tool()
def search_books(query: str) -> str:
    """Search the catalog by title or author."""
    return f"Found 3 books matching {query!r}."


async def main() -> None:
    async with Client(mcp) as client:
        result = await client.call_tool("search_books", {"query": "dune"})
        print(result.structured_content)

Ни подпроцесса, ни порта, ни байтов в передаваемых данных. Клиент и сервер — два объекта в одном процессе, и вызов всё равно проходит через настоящий протокольный уровень: search_books перечисляется, валидируется и вызывается ровно так же, как это было бы по HTTP.

Поэтому это сразу две вещи:

• Тестовый стенд. Каждый пример в этой документации проверяется именно так, а страница Тестирование строит вокруг этого весь подход.
• API для встраивания. Приложению, которое создаёт сервер, не нужен сетевой переход, чтобы вызывать его инструменты.

Streamable HTTP

Передайте строку с URL — и получите Streamable HTTP, транспорт, за которым вы развёртываете сервер:

from mcp import Client


async def main() -> None:
    async with Client("http://localhost:8000/mcp") as client:
        result = await client.list_tools()
        print([tool.name for tool in result.tools])

Это уже готовый клиент для продакшена. Client сам оборачивает URL в streamable_http_client(...) поверх httpx2.AsyncClient, настроенного так, как нужно MCP: follow_redirects=True, таймаут 30 секунд на connect/write/pool и таймаут чтения 300 секунд, потому что сервер может держать поток ответа открытым.

RuntimeError: Client must be used within an async context manager

Собственный httpx2.AsyncClient

Как только понадобится заголовок Authorization, cookie, прокси, mTLS или другой таймаут, создайте httpx2.AsyncClient сами и передайте его в streamable_http_client:

import httpx2

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


async def main() -> None:
    async with httpx2.AsyncClient(
        headers={"Authorization": "Bearer ..."},
        timeout=httpx2.Timeout(30.0, read=300.0),
        follow_redirects=True,
    ) as http_client:
        transport = streamable_http_client("http://localhost:8000/mcp", http_client=http_client)
        async with Client(transport) as client:
            result = await client.list_tools()
            print([tool.name for tool in result.tools])

Обратите внимание на две вещи:

• httpx2.AsyncClient принадлежит вам, поэтому входите в него и выходите из него вы. SDK никогда не закрывает клиент, который он не создавал.
• streamable_http_client(url, http_client=...) возвращает транспорт, а Client(transport) принимает его, как и всё остальное.

Одно замечание о TLS: httpx2 проверяет сертификаты по хранилищу доверия операционной системы (через truststore), а не по встроенному списку CA. В среде без пригодного системного хранилища CA (некоторые минимальные контейнеры) задайте стандартные переменные окружения SSL_CERT_FILE/SSL_CERT_DIR или передайте явный verify=ssl_context в свой httpx2.AsyncClient (подробности в разделе httpx и httpx-sse заменены на httpx2).

TypeError: streamable_http_client() got an unexpected keyword argument 'headers'

stdio

Сервер stdio — это подпроцесс. Клиент запускает его, пишет JSON-RPC в его stdin и читает JSON-RPC из его stdout. Именно так десктопный хост запускает сервер на вашей машине: хост — это и есть этот код плюс UI, а страница Подключение к реальному хосту показывает те же отношения со стороны хоста, в виде файла конфигурации.

Опишите процесс с помощью StdioServerParameters, превратите его в транспорт с помощью stdio_client и передайте его в Client:

from mcp import Client, StdioServerParameters
from mcp.client.stdio import stdio_client

server = StdioServerParameters(
    command="uv",
    args=["run", "server.py"],
    env={"BOOKSHOP_API_KEY": "secret"},
)


async def main() -> None:
    async with Client(stdio_client(server)) as client:
        result = await client.list_tools()
        print([tool.name for tool in result.tools])

Сам по себе объект параметров Client не принимает. StdioServerParameters — это конфигурация; stdio_client(server) — транспорт, который умеет запускать по ней процесс. Всегда оборачивайте.

Выход из блока async with заодно завершает подпроцесс: закрывает stdin, ждёт, убивает, если тот задерживается. Убирать за ним самостоятельно не нужно.

SSE

sse_client(url) из mcp.client.sse — это HTTP-транспорт, который заменил Streamable HTTP. Оборачивайте его так же, Client(sse_client("http://localhost:8000/sse")), чтобы общаться с сервером, который всё ещё на нём говорит, и не стройте на нём ничего нового.

Протокол Transport

Для Client всё перечисленное — одно и то же.

Транспорт — это любой асинхронный контекстный менеджер, который отдаёт пару потоков сообщений (read, write): формально — протокол Transport из mcp.client. Client разрешает свой аргумент по типу: объект сервера подключается внутри процесса, str превращается в streamable_http_client(url), а всё остальное используется как транспорт напрямую. Благодаря последнему правилу stdio_client(...), streamable_http_client(...) и sse_client(...) подходят в одно и то же место — и вы можете написать свой.

Итоги

• Client(mcp) (объект сервера) подключается в памяти. Используйте для тестов и для встраивания.
• Client("http://.../mcp") (URL) подключается по Streamable HTTP, транспорту для продакшена.
• Заголовки, аутентификация, прокси и таймауты задаются на httpx2.AsyncClient, который передаётся в streamable_http_client(url, http_client=...). Именованного аргумента headers= нет.
• stdio — это Client(stdio_client(StdioServerParameters(...))), и никогда не объект параметров сам по себе.
• Подпроцесс получает окружение из разрешённого списка, а не ваше; env= добавляет к нему.
• Транспорт — это всё, с чем можно написать async with x as (read, write). Client передаёт всё, что не объект сервера и не URL, прямо в этот протокол.
• Создание Client выбирает транспорт. async with его открывает.

Когда транспорт открыт, двум сторонам нужно договориться о версии протокола. Обычно думать об этом не приходится; когда всё-таки придётся, нужная страница — Версии протокола.