Push live resource-change notifications to MCP clients with the Python SDK v2 subscriptions/listen stream instead of polling.
What you'll build #
You'll take a Python MCP 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 themcp
2.0.0subscriptions/listen
method introduced in the 2026-07-28 protocol revision, which replacedresources/subscribe
. Onmcp
1.x thelisten()
API andnotify_*
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.
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
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},
}
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.
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:
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 (theasync for
exits) or abruptly (SubscriptionLost
). Neither replays missed events, and the client holds at most 1024 unconsumed events before dropping the subscription. Wraplisten()
in a loop that refetches, backs off a second, and re-listens — theclient Subscriptions pagehas 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 inspectssubscriptions/listen
requests and raisesMCPError
for URIs the caller can't read — seeserver-side Subscriptions.Scale past one process.InMemorySubscriptionBus
only reaches streams in the same process. Behind a load balancer, implement the two-methodSubscriptionBus
protocol over Redis pub/sub and pass it assubscriptions=
.Subscribe to list changes too.client.listen(tools_list_changed=True, ...)
plusctx.notify_tools_changed()
lets an agent discover tools you register at runtime withmcp.add_tool()
.Migrating from 1.x? Thev2 migration guidecovers every breaking change, including the era rules for notifications.
Sources & further reading #
Subscriptions (server side) - MCP Python SDK— py.sdk.modelcontextprotocol.io - Subscriptions (client side) - MCP Python SDK— py.sdk.modelcontextprotocol.io - Migration Guide v1 to v2 - MCP Python SDK— py.sdk.modelcontextprotocol.io - Troubleshooting - MCP Python SDK— py.sdk.modelcontextprotocol.io - Running your server - MCP Python SDK— py.sdk.modelcontextprotocol.io - mcp 2.0.0 on PyPI— pypi.org
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.