Перейти до змісту

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

Машинний переклад

Цю сторінку перекладено автоматично з англомовної документації, і основною версією є англомовна сторінка. Якщо щось читається неправильно, на сторінці Переклади пояснено, як про це повідомити.

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

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

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

У пам'яті

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

client.py
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, транспорт, за яким розгортають сервер:

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: follow_redirects=True, 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),
        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).

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-клієнти.

stdio

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

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

client.py
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, чекає, вбиває, якщо той затримується. Прибирати за ним самостійно ніколи не доведеться.

Warning

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

Сервер, якому потрібен ключ API, там його не знайде. Передайте його явно через env=; ці змінні накладаються поверх списку дозволених. Саме це й робить BOOKSHOP_API_KEY вище.

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). Усе, що не є об'єктом сервера чи URL, Client передає прямо цьому протоколу.
  • Створення Client обирає транспорт. async with його відкриває.

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