Saltar a contenido

Suscripciones

Traducción automática

Esta página se tradujo automáticamente a partir de la documentación en inglés, y la página en inglés es la versión de referencia. Si algo no se lee bien, Traducciones explica cómo avisarnos.

El catálogo de un servidor no es fijo. Las herramientas aparecen en tiempo de ejecución y el contenido detrás de la URI de un recurso cambia. Un cliente se entera a través de client.listen(...): una sola solicitud subscriptions/listen cuya respuesta es el flujo. Permanece abierta y transporta las notificaciones de cambio que el cliente pidió.

Esta página es el extremo del cliente: abrir el flujo, observarlo junto a tu flujo principal y manejar sus finales. Publicar cambios, filtrar y atender el método son la parte del servidor, que se cuenta en Suscripciones, dentro de Dentro de tu handler. Los ejemplos de aquí hablan con el servidor del tablero de sprint que se construye allí.

Observar el flujo

Una suscripción es un gestor de contexto. Al entrar se envía la solicitud, con tus argumentos nombrados como filtro de la suscripción, y se espera la confirmación del servidor, así que el flujo ya está activo cuando empieza el bloque.

client.py
from mcp import Client
from mcp.client.subscriptions import ResourceUpdated, ToolsListChanged
from mcp.types import TextResourceContents

BOARD = "board://sprint"


async def read_board(client: Client, uri: str = BOARD) -> str:
    [contents] = (await client.read_resource(uri)).contents
    assert isinstance(contents, TextResourceContents)
    return contents.text


async def follow_board(client: Client) -> None:
    async with client.listen(tools_list_changed=True, resource_subscriptions=[BOARD]) as sub:
        async for event in sub:
            match event:
                case ResourceUpdated(uri=uri):
                    print(await read_board(client, uri))
                case ToolsListChanged():
                    tools = await client.list_tools()
                    print("tools:", [tool.name for tool in tools.tools])
                case _:
                    pass  # kinds the filter did not ask for never arrive


async def main() -> None:
    async with Client("http://localhost:8000/mcp") as client:
        await follow_board(client)

La iteración produce cuatro eventos tipados: ToolsListChanged, PromptsListChanged, ResourcesListChanged y ResourceUpdated(uri=...).

Un evento dice qué cambió, nunca cómo. Por eso follow_board llama a read_resource y a list_tools: el evento es una señal para volver a pedir los datos. Lee event.uri en lugar de suponer qué recurso se movió: un filtro puede nombrar varias URI, y un servidor puede informar de un cambio en un subrecurso de una de ellas.

Los eventos duplicados que esperan a ser consumidos se funden en uno, y al volver a pedir los datos obtienes igualmente el estado actual. Solo se funden los eventos idénticos: dos ResourceUpdated para URI distintas son dos eventos.

Dos propiedades más del objeto devuelto:

  • sub.honored es el filtro que el servidor confirmó: un SubscriptionFilter con los campos que pasaste, que se leen como atributos (sub.honored.prompts_list_changed). MCPServer acepta todos los tipos que pides, así que te devuelve tu solicitud tal cual. Un servidor que admite menos tipos confirma menos, y un tipo aceptado puede no dispararse nunca. Un servidor también puede rechazar la solicitud entera en lugar de confirmarla (consulta Decidir quién puede observar en la página del servidor), lo que aparece como el error de la solicitud.
  • sub.subscription_id es el id de la solicitud de escucha, el que va estampado en cada trama de este flujo. Puede haber varias suscripciones abiertas a la vez, cada una demultiplexada por su propio id.

Observar sin bloquear

follow_board se ejecuta hasta que el servidor cierra el flujo, lo que puede no ocurrir nunca, así que por sí solo se adueña de tu programa. Los clientes reales quieren el observador junto al flujo principal: un agente llama a herramientas mientras un observador mantiene al día una caché o una interfaz.

Abre primero la suscripción, luego inicia el observador y sigue con tu trabajo.

app.py
import asyncio

from mcp import Client
from mcp.client.subscriptions import Subscription

from .tutorial003 import BOARD, read_board


async def watch(client: Client, sub: Subscription) -> None:
    async for _event in sub:
        board = await read_board(client)
        print(board)
        if "[ ]" not in board:
            return  # sprint finished: the stream closes when run_sprint leaves the block


async def run_sprint(client: Client) -> None:
    async with client.listen(resource_subscriptions=[BOARD]) as sub:
        print(await read_board(client))  # snapshot: acknowledged, so nothing after this is missed
        watcher = asyncio.create_task(watch(client, sub))
        for task in ("design", "build", "ship"):
            await client.call_tool("complete_task", {"board": "sprint", "task": task})
        await watcher  # returns once the watcher has seen the finished board


async def main() -> None:
    async with Client("http://localhost:8000/mcp") as client:
        await run_sprint(client)


if __name__ == "__main__":
    asyncio.run(main())
app.py
import trio

from mcp import Client
from mcp.client.subscriptions import Subscription

from .tutorial003 import BOARD, read_board


async def watch(client: Client, sub: Subscription) -> None:
    async for _event in sub:
        board = await read_board(client)
        print(board)
        if "[ ]" not in board:
            return  # sprint finished: the stream closes when run_sprint leaves the block


async def run_sprint(client: Client) -> None:
    async with client.listen(resource_subscriptions=[BOARD]) as sub:
        print(await read_board(client))  # snapshot: acknowledged, so nothing after this is missed
        async with trio.open_nursery() as nursery:
            nursery.start_soon(watch, client, sub)
            for task in ("design", "build", "ship"):
                await client.call_tool("complete_task", {"board": "sprint", "task": task})


async def main() -> None:
    async with Client("http://localhost:8000/mcp") as client:
        await run_sprint(client)


if __name__ == "__main__":
    trio.run(main)
app.py
import anyio

from mcp import Client
from mcp.client.subscriptions import Subscription

from .tutorial003 import BOARD, read_board


async def watch(client: Client, sub: Subscription) -> None:
    async for _event in sub:
        board = await read_board(client)
        print(board)
        if "[ ]" not in board:
            return  # sprint finished: the stream closes when run_sprint leaves the block


async def run_sprint(client: Client) -> None:
    async with client.listen(resource_subscriptions=[BOARD]) as sub:
        print(await read_board(client))  # snapshot: acknowledged, so nothing after this is missed
        async with anyio.create_task_group() as tg:
            tg.start_soon(watch, client, sub)
            for task in ("design", "build", "ship"):
                await client.call_tool("complete_task", {"board": "sprint", "task": task})


async def main() -> None:
    async with Client("http://localhost:8000/mcp") as client:
        await run_sprint(client)


if __name__ == "__main__":
    anyio.run(main)

Note

app.py importa BOARD y read_board del primer ejemplo, que este repositorio guarda como tutorial003.py. Si guardas los archivos renderizados uno junto al otro como client.py y app.py, escribe from client import BOARD, read_board en su lugar. El ejemplo watch.py de más abajo importa read_board de la misma manera.

El orden es la clave. No se reenvía nada, así que un evento publicado antes de que existiera tu flujo se pierde. Entrar en client.listen(...) espera la confirmación, así que cada cambio a partir de ese momento llega a tu observador, y la instantánea que tomas dentro del bloque no puede perderse ninguno.

Las solicitudes se ejecutan libremente junto a un flujo abierto, desde la tarea del observador o desde cualquier otra, en el mismo cliente. Como los eventos duplicados sin consumir se funden, un flujo principal ocupado puede producir una sola recarga en lugar de tres. Los eventos distintos no se funden: un filtro que nombra muchas URI encola un evento pendiente por URI.

Para dejar de observar, sal del bloque: no hay ninguna llamada unsubscribe. Cancelar la tarea propietaria del bloque lo hace por ti, y el SDK cancela la solicitud de escucha como espera el transporte: en Streamable HTTP, cerrando el flujo de esa solicitud. Un observador que se ejecuta durante toda la vida de tu app nunca vuelve por sí solo, así que cancélalo, o cancela el alcance de su grupo de tareas, al apagar.

Los flujos terminan

Un flujo termina de una de dos maneras, y ambas son flujo de control normal. Un cierre ordenado del servidor termina el async for; una caída abrupta lanza SubscriptionLost.

La diferencia es de diagnóstico, no de qué hacer después: el flujo ya no está, no se reenvió nada, y un observador al que todavía le importa vuelve a escuchar y vuelve a pedir los datos.

watch.py
import anyio

from mcp import Client
from mcp.client.subscriptions import SubscriptionLost

from .tutorial003 import read_board


async def keep_following(client: Client) -> None:
    while True:
        try:
            async with client.listen(resource_subscriptions=["board://sprint"]) as sub:
                print(await read_board(client))  # refetch: no replay across streams
                async for _event in sub:
                    print(await read_board(client))
        except SubscriptionLost:
            pass
        # Either ending means the stream is gone. Back off before re-listening:
        # a graceful close may be the server shedding load.
        await anyio.sleep(1)

Los servidores cierran flujos de forma ordenada por sus propias razones, entre ellas deshacerse de un suscriptor cuyo atraso creció demasiado, así que un final limpio no es una señal para dejar de observar. Espera un poco antes de volver a escuchar.

SubscriptionLost también tiene una causa local. El cliente retiene como máximo 1024 eventos sin consumir, y un consumidor que se atrasa tanto pierde la suscripción en lugar de crecer sin límite. Mantén corto el cuerpo del async for y haz el trabajo lento en otro sitio.

keep_following captura solo SubscriptionLost. Entrar en listen() también puede lanzar MCPError (la conexión falló o el servidor no atiende el método), TimeoutError (no llegó ninguna confirmación) y ListenNotSupportedError (una conexión anterior a 2026). Decide cuáles de ellas debe reintentar tu observador: la última nunca se recupera.

Resumen

  • Entra con async with client.listen(...); al entrar se espera la confirmación, así que no se pierde nada publicado después.
  • Itera con async for event in sub. Los eventos son señales para volver a pedir los datos, nunca cargas de datos.
  • Abre la suscripción, luego ejecuta el observador como tarea, y las llamadas a herramientas siguen fluyendo junto a él.
  • Un final limpio detiene el bucle; una caída lanza SubscriptionLost. En ambos casos: vuelve a escuchar, vuelve a pedir los datos y, antes, espera un poco.
  • Salir del bloque es darse de baja.

Publicar estos eventos, acotar el filtro y escalar más allá de un proceso son la parte del servidor: Suscripciones. Estos mismos eventos también mantienen fiable una caché del lado del cliente, y Caché es la página siguiente.