Build an MCP Server with Real-Time Resource Subscriptions Model Context Protocol (MCP) server developers can now push real-time resource-change notifications to clients using the Python SDK v2's subscriptions/listen stream, replacing the older resources/subscribe method. The tutorial, published by Mariana Souza, demonstrates building a server that emits notifications/resources/updated events from both tool calls and background tasks, with the 2026-07-28 protocol revision requiring clients to open a subscriptions/listen stream to receive updates. The example uses mcp 2.0.0 and mcp-types 2.0.0, verified on Python 3.13.5, and includes an InMemorySubscriptionBus for publishing from outside request contexts. 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: