Build an MCP Server with Real-Time Resource Subscriptions

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.
• mcp 2.0.0 from PyPI (released 2026-07-28). This tutorial is v2-only: it relies on the 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.

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.

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 (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 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.
• 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 covers every breaking change, including the era rules for notifications.