Client-Transporte

Jeder Client spricht mit seinem Server über einen Transport: das, was die Nachrichten tatsächlich befördert.

Du konfigurierst nie einen separat. Client nimmt ein einziges positionales Argument und leitet den Transport aus dessen Typ ab.

Die Server-Seite jedes Transports (was mcp.run() tut und was du bereitstellst) steht in Den Server betreiben.

Streamable HTTP

Übergib einen URL-String und du bekommst Streamable HTTP – den Transport, hinter dem du bereitstellst und zu dem du zuerst greifen solltest:

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

Das ist der ganze Produktions-Client. Client packt die URL für dich in streamable_http_client(...), auf Basis eines httpx2.AsyncClient, der so konfiguriert ist, wie MCP es braucht: ein Timeout von 30 Sekunden für connect/write/pool und ein Read-Timeout von 300 Sekunden, weil der Server einen Response-Stream offen halten kann.

RuntimeError: Client must be used within an async context manager

Einen eigenen httpx2.AsyncClient mitbringen

Sobald du einen Authorization-Header, ein Cookie, einen Proxy, mTLS oder ein anderes Timeout brauchst, baust du den httpx2.AsyncClient selbst und übergibst ihn an 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])

Zwei Dinge fallen auf:

• Der httpx2.AsyncClient gehört dir, also betrittst und verlässt du ihn. Das SDK schließt nie einen Client, den es nicht selbst erzeugt hat.
• streamable_http_client(url, http_client=...) gibt einen Transport zurück, und Client(transport) nimmt ihn an wie alles andere auch.

Eine Anmerkung zu TLS: httpx2 prüft Zertifikate gegen den Trust Store des Betriebssystems (über truststore), nicht gegen eine mitgelieferte CA-Liste. In einer Umgebung ohne nutzbaren System-CA-Store (manche minimalen Container) setzt du die Standard-Umgebungsvariablen SSL_CERT_FILE/SSL_CERT_DIR oder übergibst deinem httpx2.AsyncClient ein explizites verify=ssl_context (Hintergrund in httpx und httpx-sse durch httpx2 ersetzt).

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

Redirects

Der Transport verbindet sich mit der URL, die du ihm gegeben hast, und nur mit diesem Origin.

• Einem 307/308-Redirect, der auf demselben Schema, Host und Port bleibt, wird gefolgt, ebenso http:// → https:// auf demselben Host. Das deckt den üblichen Trailing-Slash-Redirect /mcp → /mcp/ ab.

• Einem Redirect irgendwo anders hin wird nicht gefolgt. Der Aufruf schlägt fehl mit:

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

  Ist diese URL der Server, den du meintest, trag sie in deine Konfiguration ein. Wenn nicht, ist der Server oder ein Proxy davor falsch konfiguriert.

Das gilt für jeden httpx2.AsyncClient, den du übergibst: Seine Einstellung follow_redirects wird für MCP-Requests nicht herangezogen, in keine der beiden Richtungen. Die OAuth-Provider des SDK wenden dieselbe Regel auf ihre eigenen Requests an.

stdio

Ein stdio-Server ist ein Subprozess. Der Client startet ihn, schreibt JSON-RPC in seine stdin und liest JSON-RPC aus seiner stdout. So betreibt ein Desktop-Host einen Server auf deinem Rechner: Ein Host ist dieser Code plus eine UI, und Mit einem echten Host verbinden zeigt dieselbe Beziehung von der Seite des Hosts, als Konfigurationsdatei.

Beschreibe den Prozess mit StdioServerParameters und übergib das Objekt an 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])

Beim Eintreten in den Block wird der Prozess gestartet. Beim Verlassen wird der Subprozess beendet: stdin schließen, warten, abschießen, falls er hängen bleibt. Du räumst ihn nie selbst auf.

Die stderr des Kindprozesses landet in deiner. Um sie woandershin zu leiten, baust du den Transport selbst mit stdio_client (aus mcp) und übergibst stattdessen diesen: Client(stdio_client(server, errlog=log_file)).

Im Speicher

In einem Test gibt es nichts bereitzustellen und nichts zu starten. Übergib das Server-Objekt selbst:

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)

Kein Subprozess, kein Port, keine Bytes auf einer Leitung. Client und Server sind zwei Objekte im selben Prozess, und der Aufruf läuft trotzdem durch die echte Protokollschicht: search_books wird genau so aufgelistet, validiert und aufgerufen, wie es über HTTP geschähe. Testen baut das ganze Muster darauf auf.

Dieselbe Form dient zugleich als Embedding-API: Eine Anwendung, die den Server selbst erzeugt, kann dessen Tools ohne Netzwerk-Hop aufrufen.

SSE

sse_client(url) aus mcp.client.sse ist der HTTP-Transport, den Streamable HTTP abgelöst hat. Pack ihn genauso ein, Client(sse_client("http://localhost:8000/sse")), um mit einem Server zu sprechen, der ihn noch verwendet – und bau nichts Neues darauf.

Das Transport-Protokoll

Für Client ist alles oben Genannte dasselbe.

Ein Transport ist ein beliebiger asynchroner Kontextmanager, der ein (read, write)-Paar von Nachrichten-Streams liefert: formal das Transport-Protokoll in mcp.client. Client löst sein Argument nach Typ auf: Ein str wird zu streamable_http_client(url), ein StdioServerParameters wird zu stdio_client(params), ein Server-Objekt verbindet im Prozess, und alles andere wird direkt als Transport betreten. Diese letzte Regel ist der Grund, warum stdio_client(...), streamable_http_client(...) und sse_client(...) alle in denselben Platz passen – und warum du deinen eigenen schreiben kannst.

Zusammenfassung

• Client("http://.../mcp") (eine URL) verbindet über Streamable HTTP, den Produktions-Transport.
• Header, Auth, Proxys und Timeouts gehören auf einen httpx2.AsyncClient, den du an streamable_http_client(url, http_client=...) übergibst. Es gibt kein Keyword headers=.
• Redirects wird nur innerhalb des eigenen Origins der URL gefolgt (ein Trailing-Slash-307/308), plus http→https auf demselben Host. Alles andere schlägt mit Redirect to … not followed fehl; konfiguriere die endgültige URL.
• stdio ist Client(StdioServerParameters(...)). Pack es nur dann selbst in stdio_client(...) ein, wenn du die stderr des Kindprozesses umleiten willst.
• Der Subprozess bekommt eine Umgebung per Allow-List, nicht deine; env= ergänzt sie.
• Client(mcp) (das Server-Objekt) verbindet im Speicher. Nutze es in Tests oder um einen Server in die Anwendung einzubetten, die ihn gebaut hat.
• Ein Transport ist alles, womit du async with x as (read, write) schreiben kannst. Alles, was weder Server-Objekt noch URL noch StdioServerParameters ist, reicht Client direkt an dieses Protokoll weiter.
• Das Erzeugen eines Client wählt den Transport. async with öffnet ihn.

Sobald der Transport offen ist, müssen sich beide Seiten auf eine Protokollversion einigen. Normalerweise denkst du nie darüber nach; wenn doch, ist Protokollversionen die richtige Seite.