# Build an MCP Server with Real-Time Resource Subscriptions

> Source: <https://sourcefeed.dev/a/build-an-mcp-server-with-real-time-resource-subscriptions>
> Published: 2026-08-23 11:39:43+00:00

# Build an MCP Server with Real-Time Resource Subscriptions

Push live resource-change notifications to MCP clients with the Python SDK v2 subscriptions/listen stream instead of polling.

[Mariana Souza](https://sourcefeed.dev/u/mariana_souza)

## What you'll build

You'll take a Python [MCP](https://modelcontextprotocol.io/) server that exposes deployment state as resources and extend it so connected clients get pushed `notifications/resources/updated`

events the instant something changes — from a tool call *and* from a background task — instead of re-reading resources on a timer. You'll finish with a running HTTP server, a subscriber client that refetches on every event, and the exact wire frames to prove it.

## Prerequisites

**Python 3.10+**(verified on 3.13.5). The SDK is written against anyio, so asyncio or trio both work.from PyPI (released 2026-07-28). This tutorial is v2-only: it relies on the`mcp`

2.0.0`subscriptions/listen`

method introduced in the 2026-07-28 protocol revision, which replaced`resources/subscribe`

. On`mcp`

1.x the`listen()`

API and`notify_*`

helpers don't exist.- macOS or Linux shell. Windows works the same with
`.venv\Scripts\`

paths. - No API keys or accounts — everything runs on localhost.

One thing to know before you start: in the 2026-07-28 spec, change notifications reach a client **only** over a `subscriptions/listen`

stream the client opened. The old per-session helpers (`ctx.session.send_resource_updated(uri)`

) are silently dropped on a 2026-era connection. If you've built subscriptions on the 2025 protocol, the server-side API below is the one you migrate to.

## 1. Set up the project

```
mkdir deploy-board && cd deploy-board
python3 -m venv .venv && source .venv/bin/activate
pip install "mcp[cli]"
pip list | grep -E "^mcp"
```

You should see `mcp 2.0.0`

and `mcp-types 2.0.0`

. The `[cli]`

extra adds the `mcp dev`

Inspector launcher; you won't need it here but it's useful later.

## 2. Write the server

Create `server.py`

. The pieces that matter are the `InMemorySubscriptionBus`

you construct yourself (so code outside a request can publish on it), the `notify_resource_updated()`

call in the tool, and the background `health_poller`

publishing from the lifespan.

``` python
import asyncio
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from mcp.server import MCPServer
from mcp.server.mcpserver import Context
from mcp.server.subscriptions import InMemorySubscriptionBus, ResourceUpdated

# Fake deployment state. Swap for a database, Redis, or a CI webhook.
DEPLOYS: dict[str, dict[str, str | int]] = {
    "api": {"version": "1.4.2", "status": "healthy", "restarts": 0},
    "worker": {"version": "0.9.0", "status": "healthy", "restarts": 0},
}

# Hold the bus yourself so code outside a request (a background task,
# a webhook handler) can publish on it too.
bus = InMemorySubscriptionBus()

async def health_poller() -> None:
    """Simulate an external system: bump a counter every 3s and publish."""
    while True:
        await asyncio.sleep(3)
        DEPLOYS["worker"]["restarts"] = int(DEPLOYS["worker"]["restarts"]) + 1
        await bus.publish(ResourceUpdated(uri="deploy://worker"))

@asynccontextmanager
async def lifespan(server: MCPServer) -> AsyncIterator[dict]:
    task = asyncio.create_task(health_poller())
    try:
        yield {}
    finally:
        task.cancel()

mcp = MCPServer("Deploy Board", lifespan=lifespan, subscriptions=bus)

@mcp.resource("deploy://{service}")
def deploy(service: str) -> str:
    """Current deployment state of one service."""
    d = DEPLOYS[service]
    return f"{service} v{d['version']} status={d['status']} restarts={d['restarts']}"

@mcp.tool()
async def set_status(service: str, status: str, ctx: Context) -> str:
    """Mark a service healthy, degraded, or down."""
    DEPLOYS[service]["status"] = status
    await ctx.notify_resource_updated(f"deploy://{service}")
    return f"{service} is now {status}"

if __name__ == "__main__":
    mcp.run(transport="streamable-http", port=8000)
```

Why two publish paths: `ctx.notify_resource_updated()`

is the one-liner for changes your own handler makes. `bus.publish(ResourceUpdated(uri=...))`

is for changes that originate elsewhere — a poller, a webhook, a queue consumer — where there's no request context. `MCPServer`

builds a bus internally if you pass nothing, but doesn't expose it, which is why you construct one and pass `subscriptions=bus`

.

The SDK serves `subscriptions/listen`

for you: acknowledgment as the first frame, subscription id stamped on every frame, per-stream filtering. Publishing with no subscribers is a no-op.

Note the lifespan uses `asyncio.create_task`

, which pins this server to asyncio. If you run under trio, start the poller in a task group instead.

Start it:

```
python server.py
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
```

Leave it running and open a second terminal.

## 3. Write the subscriber client

Create `client.py`

. `client.listen()`

sends the `subscriptions/listen`

request and waits for the server's acknowledgment before the `async with`

body runs, so the snapshot you take inside the block can't miss an update.

``` python
import asyncio

from mcp import Client
from mcp.client.subscriptions import ResourceUpdated
from mcp.types import TextResourceContents

URIS = ["deploy://api", "deploy://worker"]

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

async def main() -> None:
    async with Client("http://127.0.0.1:8000/mcp") as client:
        async with client.listen(resource_subscriptions=URIS) as sub:
            print("subscribed:", sub.honored.resource_subscriptions)
            for uri in URIS:
                print("snapshot:", await read(client, uri))
            async for event in sub:
                if isinstance(event, ResourceUpdated):
                    print("updated:", await read(client, event.uri))

if __name__ == "__main__":
    asyncio.run(main())
```

An event is a cue, not a payload — the frame carries the URI and nothing else, so the client refetches. Read `event.uri`

rather than assuming which resource moved; one filter can name many URIs. Leaving the `async with`

block is the unsubscribe; there's no explicit call.

Run it:

```
source .venv/bin/activate
python client.py
```

## 4. Trigger a change from a tool call

Create `poke.py`

— a second client that calls `set_status`

, the way an LLM host would:

``` php
import asyncio

from mcp import Client

async def main() -> None:
    async with Client("http://127.0.0.1:8000/mcp") as client:
        result = await client.call_tool("set_status", {"service": "api", "status": "degraded"})
        print(result.content[0].text)

if __name__ == "__main__":
    asyncio.run(main())
```

In a third terminal:

```
source .venv/bin/activate
python poke.py
```

## Verify it works

`poke.py`

prints:

```
api is now degraded
```

The `client.py`

terminal shows the acknowledged filter, the two snapshots, a stream of `worker`

updates every ~3 s from the background poller, and the `api`

update the moment `poke.py`

ran:

```
subscribed: ['deploy://api', 'deploy://worker']
snapshot: api v1.4.2 status=healthy restarts=0
snapshot: worker v0.9.0 status=healthy restarts=1
updated: worker v0.9.0 status=healthy restarts=2
updated: worker v0.9.0 status=healthy restarts=3
updated: api v1.4.2 status=degraded restarts=0
updated: worker v0.9.0 status=healthy restarts=4
```

Both publish paths delivered — the tool's `ctx.notify_resource_updated()`

and the lifespan task's `bus.publish()`

— and only to the URIs this stream asked for. On the wire, the stream looks like this:

```
{"method": "notifications/subscriptions/acknowledged",
 "params": {"notifications": {"resourceSubscriptions": ["deploy://api", "deploy://worker"]},
            "_meta": {"io.modelcontextprotocol/subscriptionId": "listen-1"}}}

{"method": "notifications/resources/updated",
 "params": {"uri": "deploy://api", "_meta": {"io.modelcontextprotocol/subscriptionId": "listen-1"}}}
```

Stop `client.py`

with Ctrl+C; the server logs nothing special, because closing the listen request's stream is how a client unsubscribes.

## Troubleshooting

** mcp.shared.exceptions.MCPError: Method not found** when a client calls

`client.subscribe_resource(uri)`

. You're also seeing `MCPDeprecationWarning: resources/subscribe is removed as of 2026-07-28; use Client.listen() instead.`

A 2026-07-28 server answers `resources/subscribe`

with `-32601`

. Replace the call with `async with client.listen(resource_subscriptions=[uri])`

. Keep `subscribe_resource()`

only for talking to 2025-era servers, and filter the warning there.** mcp.client.subscriptions.ListenNotSupportedError: subscriptions/listen is not available at protocol version '2025-11-25'; it requires 2026-07-28.** The client negotiated an older protocol, usually because you passed

`mode="legacy"`

to `Client(...)`

or the server is on `mcp`

1.x. Drop `mode="legacy"`

, or upgrade the server. This one never heals on retry, so don't wrap it in a reconnect loop.**Updates never arrive, no error anywhere.** Your server still calls `ctx.session.send_resource_updated(uri)`

. On a 2026-07-28 connection that helper is dropped with a debug log — it pushes onto a standalone channel that `subscriptions/listen`

streams don't read. Switch to `await ctx.notify_resource_updated(uri)`

(or `bus.publish(ResourceUpdated(uri=...))`

). Also check the URI string matches exactly: `MCPServer`

compares as exact strings, so a subscription to `deploy://api`

hears nothing about `deploy://api/pods`

.

** mcp.shared.exceptions.MCPError: Server returned an error response** right after deploying behind a real hostname. The server log shows

`WARNING mcp.server.transport_security: Invalid Host header: <your-host>`

(HTTP 421). DNS-rebinding protection is on by default and accepts only localhost `Host`

headers. Pass `transport_security=TransportSecuritySettings(allowed_hosts=["mcp.example.com", "mcp.example.com:*"], allowed_origins=[...])`

(from `mcp.server.transport_security`

) to `mcp.run(...)`

or `mcp.streamable_http_app(...)`

.## Next steps

**Reconnect logic.** A stream ends gracefully (the`async for`

exits) or abruptly (`SubscriptionLost`

). Neither replays missed events, and the client holds at most 1024 unconsumed events before dropping the subscription. Wrap`listen()`

in a loop that refetches, backs off a second, and re-listens — the[client Subscriptions page](https://py.sdk.modelcontextprotocol.io/client/subscriptions/)has the pattern.**Gate who may watch.** By default any caller can listen on any URI, including ones your read handler would refuse. Add a middleware that inspects`subscriptions/listen`

requests and raises`MCPError`

for URIs the caller can't read — see[server-side Subscriptions](https://py.sdk.modelcontextprotocol.io/handlers/subscriptions/).**Scale past one process.**`InMemorySubscriptionBus`

only reaches streams in the same process. Behind a load balancer, implement the two-method`SubscriptionBus`

protocol over Redis pub/sub and pass it as`subscriptions=`

.**Subscribe to list changes too.**`client.listen(tools_list_changed=True, ...)`

plus`ctx.notify_tools_changed()`

lets an agent discover tools you register at runtime with`mcp.add_tool()`

.**Migrating from 1.x?** The[v2 migration guide](https://py.sdk.modelcontextprotocol.io/migration/)covers every breaking change, including the era rules for notifications.

## Sources & further reading

-
[Subscriptions (server side) - MCP Python SDK](https://py.sdk.modelcontextprotocol.io/handlers/subscriptions/)— py.sdk.modelcontextprotocol.io -
[Subscriptions (client side) - MCP Python SDK](https://py.sdk.modelcontextprotocol.io/client/subscriptions/)— py.sdk.modelcontextprotocol.io -
[Migration Guide v1 to v2 - MCP Python SDK](https://py.sdk.modelcontextprotocol.io/migration/)— py.sdk.modelcontextprotocol.io -
[Troubleshooting - MCP Python SDK](https://py.sdk.modelcontextprotocol.io/troubleshooting/)— py.sdk.modelcontextprotocol.io -
[Running your server - MCP Python SDK](https://py.sdk.modelcontextprotocol.io/run/)— py.sdk.modelcontextprotocol.io -
[mcp 2.0.0 on PyPI](https://pypi.org/project/mcp/)— pypi.org

[Mariana Souza](https://sourcefeed.dev/u/mariana_souza)· Senior Editor

Mariana covers the fast-moving world of machine learning and generative AI, with a particular focus on how these technologies are reshaping development workflows. When she isn't stress-testing the latest foundation models, she's usually at a local hackathon.

## Discussion 0

No comments yet

Be the first to weigh in.