Transportes del cliente

Cada Client habla con su servidor a través de un transporte: lo que realmente lleva los mensajes.

Nunca configuras uno por separado. Client recibe un único argumento posicional y deduce el transporte a partir de su tipo.

El lado del servidor de cada uno (lo que hace mcp.run() y lo que despliegas) está en Ejecutar el servidor.

En memoria

Pasa el propio objeto del servidor:

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)

Sin subproceso, sin puerto, sin bytes por ningún canal. El cliente y el servidor son dos objetos en el mismo proceso, y aun así la llamada pasa por la capa real del protocolo: search_books se lista, se valida y se invoca exactamente igual que por HTTP.

Eso lo convierte en dos cosas a la vez:

• Un arnés de pruebas. Todos los ejemplos de esta documentación se ejercitan así, y la página Pruebas construye todo el patrón en torno a ello.
• Una API de integración. Una aplicación que construye el servidor no necesita un salto de red para llamar a sus herramientas.

Streamable HTTP

Pasa una cadena con una URL y obtienes Streamable HTTP, el transporte con el que despliegas:

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

Ese es todo el cliente de producción. Client envuelve la URL en streamable_http_client(...) por ti, encima de un httpx2.AsyncClient configurado como MCP necesita: follow_redirects=True, un timeout de 30 segundos para connect/write/pool y un timeout de lectura de 300 segundos, porque el servidor puede mantener abierto un flujo de respuesta.

RuntimeError: Client must be used within an async context manager

Trae tu propio httpx2.AsyncClient

En cuanto necesites un encabezado Authorization, una cookie, un proxy, mTLS o un timeout distinto, construye tú mismo el httpx2.AsyncClient y entrégaselo a 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])

Dos cosas que notar:

• El httpx2.AsyncClient es tuyo, así que tú entras y sales de él. El SDK nunca cierra un cliente que no creó.
• streamable_http_client(url, http_client=...) devuelve un transporte, y Client(transport) lo acepta como cualquier otra cosa.

Una nota sobre TLS: httpx2 verifica los certificados contra el almacén de confianza del sistema operativo (mediante truststore), no contra una lista de CA incluida. En un entorno sin un almacén de CA del sistema utilizable (algunos contenedores mínimos), configura las variables de entorno estándar SSL_CERT_FILE/SSL_CERT_DIR o pasa un verify=ssl_context explícito a tu httpx2.AsyncClient (el contexto está en httpx y httpx-sse sustituidos por httpx2).

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

stdio

Un servidor stdio es un subproceso. El cliente lo lanza, escribe JSON-RPC en su stdin y lee JSON-RPC de su stdout. Así es como un host de escritorio ejecuta un servidor en tu máquina: un host es este código más una interfaz de usuario, y Conectar a un host real es la misma relación vista desde el lado del host, como archivo de configuración.

Describe el proceso con StdioServerParameters, conviértelo en un transporte con stdio_client y entrega eso a 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 no acepta el objeto de parámetros por sí solo. StdioServerParameters es configuración; stdio_client(server) es el transporte que sabe lanzar un proceso a partir de ella. Envuélvelo siempre.

Salir del bloque async with también cierra el subproceso: cierra stdin, espera y lo mata si se queda. Nunca lo limpias tú.

SSE

sse_client(url), de mcp.client.sse, es el transporte HTTP al que reemplazó Streamable HTTP. Envuélvelo igual, Client(sse_client("http://localhost:8000/sse")), para hablar con un servidor que todavía lo usa, y no construyas nada nuevo sobre él.

El protocolo Transport

Para Client, todo lo anterior es lo mismo.

Un transporte es cualquier gestor de contexto asíncrono que produce un par (read, write) de flujos de mensajes: formalmente, el protocolo Transport de mcp.client. Client resuelve su argumento por tipo: un objeto de servidor se conecta dentro del proceso, un str se convierte en streamable_http_client(url) y cualquier otra cosa se entra directamente como transporte. Esa última regla es la razón por la que stdio_client(...), streamable_http_client(...) y sse_client(...) encajan todos en el mismo hueco, y por la que puedes escribir el tuyo.

Resumen

• Client(mcp) (el objeto del servidor) se conecta en memoria. Úsalo para pruebas y para integración.
• Client("http://.../mcp") (una URL) se conecta por Streamable HTTP, el transporte de producción.
• Los encabezados, la autenticación, los proxies y los timeouts van en un httpx2.AsyncClient que pasas a streamable_http_client(url, http_client=...). No existe el argumento nombrado headers=.
• stdio es Client(stdio_client(StdioServerParameters(...))), nunca el objeto de parámetros solo.
• El subproceso recibe un entorno con lista de permitidos, no el tuyo; env= se añade a él.
• Un transporte es cualquier cosa con la que puedas hacer async with x as (read, write). Client entrega directamente a ese protocolo todo lo que no sea un objeto de servidor ni una URL.
• Construir un Client elige el transporte. async with lo abre.

Una vez abierto el transporte, los dos lados tienen que acordar una versión del protocolo. Normalmente nunca piensas en ello; cuando lo hagas, Versiones del protocolo es la página.