Транспорти клієнта

Кожен Client спілкується зі своїм сервером через транспорт — те, що власне й переносить повідомлення.

Окремо його налаштовувати не доводиться. Client приймає один позиційний аргумент і визначає транспорт за його типом.

Серверний бік кожного з них (що робить mcp.run() і що ви розгортаєте) описано на сторінці Запуск сервера.

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: 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),
    ) 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) приймає його, як і будь-що інше.

Залиште timeout=. Це той самий тайм-аут, який використовує власний клієнт SDK (30 секунд, 300 на читання); httpx2.AsyncClient, створений без нього, отримує типовий для httpx2 5-секундний тайм-аут, і виклик інструмента, що триває довше, завершується помилкою тайм-ауту читання.

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

Більші SSE-події

Передайте max_sse_event_size, якщо сервер надсилає великий результат інструмента чи сповіщення в одній SSE-події:

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


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

За замовчуванням ліміт становить 1 МіБ на подію й вимірюється в байтах до розбору події. Ліміт діє для POST-відповідей, GET-потоку й відновлених потоків. Завелика подія в POST-відповіді чи відновленому потоці завершує цей запит помилкою SSE. У фоновому GET-потоці клієнт записує помилку в лог і пробує відкрити потік знову. Задайте max_sse_event_size=None, щоб вимкнути обмеження, якщо довіряєте серверу й потрібні більші події. JSON-відповідей це не стосується. Якщо ви використовуєте ClientSessionGroup, задайте той самий параметр у StreamableHttpParameters.

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

Перенаправлення

Транспорт під'єднується до 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 застосовують те саме правило до власних запитів.

stdio

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

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

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)).

У пам'яті

У тесті нема чого розгортати й нема чого запускати. Передайте сам об'єкт сервера:

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= немає.
• Щоб змінити ліміт у байтах для кожної SSE-події, використовуйте streamable_http_client(url, max_sse_event_size=...).
• Перенаправлення виконуються лише в межах власного origin цього 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). Усе, що не є об'єктом сервера, URL чи StdioServerParameters, Client передає прямо цьому протоколу.
• Створення Client обирає транспорт. async with його відкриває.

Щойно транспорт відкрито, обидві сторони мають домовитися про версію протоколу. Зазвичай про це й не згадуєте; а коли все ж доведеться — є сторінка Версії протоколу.