# Subscriptions

A server's catalog is not fixed. Tools appear at runtime, and the content behind a resource URI changes. A client hears about it through `client.listen(...)`: one `subscriptions/listen` request whose response *is* the stream. It stays open and carries the change notifications the client asked for.

This page is the client end: opening the stream, watching it beside your main flow, and handling its endings. Publishing changes, filtering, and serving the method are the server's side of the story, told in **[Subscriptions](https://py.sdk.modelcontextprotocol.io/handlers/subscriptions/index.md)** under *Inside your handler*. The examples here talk to the sprint-board server built there.

## Watching the stream

A subscription is one context manager. Entering it sends the request, with your keyword arguments as the subscription filter, and waits for the server's acknowledgment, so the stream is live by the time the block starts.

```python title="client.py" hl_lines="15 18 28"
# docs_src/subscriptions/tutorial003.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)
```

Iteration yields four typed events: `ToolsListChanged`, `PromptsListChanged`, `ResourcesListChanged`, and `ResourceUpdated(uri=...)`.

An event says *what* changed, never *how*. That is why `follow_board` calls `read_resource` and `list_tools`: the event is a cue to refetch. Read `event.uri` rather than assuming which resource moved: a filter can name several URIs, and a server may report a change on a sub-resource of one of them.

Duplicate events waiting to be consumed collapse into one, and refetching still gets you the current state. Only identical events collapse: two `ResourceUpdated` for different URIs are two events.

Two more properties of the handle:

* `sub.honored` is the filter the server acknowledged: a `SubscriptionFilter` with the fields you passed, read as attributes (`sub.honored.prompts_list_changed`). `MCPServer` honors every kind you ask for, so it echoes your request back. A server that supports fewer kinds acknowledges less, and an honored kind may still never fire. A server may also refuse the whole request rather than acknowledge it (see [Deciding who may watch](https://py.sdk.modelcontextprotocol.io/handlers/subscriptions/index.md#deciding-who-may-watch) on the server page), which surfaces as the request's error.
* `sub.subscription_id` is the listen request's id, the one stamped on every frame of this stream. Several subscriptions can be open at once, each demultiplexed by its own id.

## Watching without blocking

`follow_board` runs until the server closes the stream, which may be never, so on its own it owns your program. Real clients want the watcher *beside* the main flow: an agent calls tools while a watcher keeps a cache or a UI current.

Open the subscription first, then start the watcher and get on with your work.

=== "asyncio"

    ```python title="app.py" hl_lines="18 20"
    # docs_src/subscriptions/tutorial004_asyncio.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())
    ```

=== "trio"

    ```python title="app.py" hl_lines="18 21"
    # docs_src/subscriptions/tutorial004_trio.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)
    ```

=== "anyio"

    ```python title="app.py" hl_lines="18 21"
    # docs_src/subscriptions/tutorial004_anyio.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` imports `BOARD` and `read_board` from the first example, which this repo stores as
    `tutorial003.py`. If you save the rendered files side by side as `client.py` and `app.py`,
    write `from client import BOARD, read_board` instead. The `watch.py` example further down
    imports `read_board` the same way.

The order is the point. Nothing is replayed, so an event published before your stream existed is missed. Entering `client.listen(...)` waits for the acknowledgment, so every change from that moment on reaches your watcher, and the snapshot you take inside the block cannot miss one.

Requests run freely beside an open stream, from the watcher task or any other, on the same client. Because *duplicate* unconsumed events coalesce, a busy main flow may produce one refetch rather than three. Events that differ do not coalesce: a filter naming many URIs queues one pending event per URI.

To stop watching, leave the block: there is no `unsubscribe` call. Cancelling the task that owns the block does that for you, and the SDK cancels the listen request the way the transport expects: over streamable HTTP, by closing that request's stream. A watcher that runs for the life of your app never returns on its own, so cancel it, or its task group's scope, at shutdown.

## Streams end

A stream ends in one of two ways, both ordinary control flow. A graceful server close ends the `async for`; an abrupt drop raises `SubscriptionLost`.

The difference is diagnostic, not a difference in what to do next: the stream is gone, nothing was replayed, and a watcher that still cares re-listens and refetches.

```python title="watch.py" hl_lines="16 20"
# docs_src/subscriptions/tutorial005.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)
```

Servers close streams gracefully for their own reasons, including shedding a subscriber whose backlog grew too large, so a clean end is not a signal to stop watching. Back off before re-listening.

`SubscriptionLost` has one local cause too. The client holds at most 1024 unconsumed events, and a consumer that falls that far behind loses the subscription rather than grow without bound. Keep the body of the `async for` short and do slow work elsewhere.

`keep_following` catches only `SubscriptionLost`. Entering `listen()` can also raise `MCPError` (the connection failed, or the server does not serve the method), `TimeoutError` (no acknowledgment arrived), and `ListenNotSupportedError` (a pre-2026 connection). Decide which of those your watcher should retry: the last never heals.

## Recap

* Enter `async with client.listen(...)`; entering waits for the acknowledgment, so nothing published after it is missed.
* Iterate with `async for event in sub`. Events are cues to refetch, never payloads.
* Open the subscription, then run the watcher as a task, and tool calls keep flowing beside it.
* A clean end stops the loop; a drop raises `SubscriptionLost`. Either way: re-listen, refetch, back off first.
* Leaving the block is the unsubscribe.

Publishing these events, narrowing the filter, and scaling past one process are the server's story: **[Subscriptions](https://py.sdk.modelcontextprotocol.io/handlers/subscriptions/index.md)**. These same events also keep a client-side cache honest, and **[Caching](https://py.sdk.modelcontextprotocol.io/client/caching/index.md)** is the next page.