Перейти к содержанию

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

Машинный перевод

Эта страница переведена с английской документации автоматически, и основной версией остаётся английская страница. Если что-то читается неправильно, на странице Переводы объясняется, как об этом сообщить.

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

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

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

Streamable HTTP

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

client.py
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: таймаут 30 секунд на connect/write/pool и таймаут чтения 300 секунд, потому что сервер может держать поток ответа открытым.

Check

Созданный Client не подключён. Конструктор только выбирает транспорт; открывает его async with. Обратитесь к соединению до входа в блок — и SDK сообщит об этом:

RuntimeError: Client must be used within an async context manager

Когда вы написали Client("http://..."), ничего не разрешалось, не загружалось и не запускалось. Эта строка ничего не стоит.

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

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

client.py
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),
    ) 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).

Warning

Раньше streamable_http_client принимал headers= и timeout= напрямую. Больше не принимает: его единственные параметры — url, http_client и terminate_on_close. Напишите по привычке headers= — и получите:

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

Всё, что относится к HTTP, теперь живёт в одном httpx2.AsyncClient, который вы передаёте.

Info

httpx2 сохраняет привычный API httpx, так что, если вы знаете httpx, вы уже умеете делать здесь аутентификацию, прокси, хуки событий, повторные попытки и ограничения соединений. SDK ничего не добавляет сверху и ничего не убирает, кроме обработки редиректов. Здесь же подключается OAuth: httpx2.AsyncClient(auth=OAuthClientProvider(...)). Весь этот сценарий — на странице OAuth-клиенты.

Редиректы

Транспорт подключается к URL, который вы ему передали, и ни к какому другому источнику (origin).

  • Редирект 307/308, который остаётся на той же схеме, хосте и порту, выполняется; то же касается перехода http:// → https:// на том же хосте. Это покрывает обычный редирект с добавлением косой черты /mcp → /mcp/.

  • Редирект куда-либо ещё не выполняется. Вызов завершается ошибкой:

    MCPError: Redirect to https://other.example.com/mcp not followed; use that URL as the endpoint if it is the intended server

    Если этот URL и есть нужный сервер, укажите его в конфигурации. Если нет — сервер или прокси перед ним настроен неправильно.

Это верно для любого httpx2.AsyncClient, который вы передаёте: его настройка follow_redirects для MCP-запросов не учитывается — ни в одну, ни в другую сторону. OAuth-провайдеры SDK применяют то же правило к своим собственным запросам.

Tip

Сообщение Redirect to http://… not followed: it would downgrade this HTTPS endpoint to plain HTTP означает, что сервер стоит за прокси с терминацией TLS, о котором не знает, и выдаёт редиректы на http://. Это исправляется на сервере (Развёртывание и масштабирование) либо использованием ровно того URL https://…/, который предлагает сообщение.

stdio

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

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

client.py
from mcp import Client, StdioServerParameters

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


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

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

stderr дочернего процесса идёт в ваш. Чтобы направить его куда-то ещё, соберите транспорт сами с помощью stdio_client (из mcp) и передайте вместо этого его: Client(stdio_client(server, errlog=log_file)).

Warning

Дочерний процесс не наследует ваше окружение. Он получает минимальный разрешённый список (HOME, LOGNAME, PATH, SHELL, TERM и USER в POSIX), чтобы ничего чувствительного не утекло в процесс, который, возможно, написали не вы.

Сервер, которому нужен API-ключ, там его не найдёт. Передайте его явно через env=; эти переменные добавляются поверх разрешённого списка. Именно это делает BOOKSHOP_API_KEY выше.

В памяти

В тесте нечего развёртывать и нечего запускать. Передайте сам объект сервера:

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 для встраивания: приложение, которое само создаёт сервер, может вызывать его инструменты без обращения к сети.

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), StdioServerParameters — в stdio_client(params), объект сервера подключается внутри процесса, а всё остальное открывается напрямую как транспорт. Благодаря последнему правилу stdio_client(...), streamable_http_client(...) и sse_client(...) встают в одно и то же место — и поэтому же можно написать свой собственный.

Итоги

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

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