Real-Time Updates and Notifications

Notifications let a server send something the client did not individually ask for — progress on a slow call, or word that a resource changed. What they do not do is let the server start a conversation. The specification is blunt about it: servers MUST NOT initiate JSON-RPC requests. Every notification is scoped to a request the client made.

That leaves exactly two ways a notification reaches a client, and picking the right one is the whole lesson.

Notification Types

The subscription path is the one people reach for and get wrong. subscriptions/listen is an ordinary request — it just happens to have a response that stays open. Its state belongs to that request, not to the connection underneath, so if the channel drops the client re-issues the request rather than expecting your server to remember anything.

Type	What it's for	What goes wrong without it
Progress	A tool that takes longer than a user will sit still for	The host cannot distinguish "working" from "hung"
Resource updated	Content the host may have cached has changed	The model answers from a stale copy and sounds confident about it
System alerts	Degraded state the host should surface	Failures stay invisible until a tool call fails outright

The important property: a notification carries no reply. Nothing waits on it, and nothing retries if it is lost. That makes them safe to send often, and unsafe to depend on for correctness — never let a notification be the only way a client learns something it needs.

Sending Notifications

from mcp.server.mcpserver import Context


@mcp.tool()
async def reindex(collection: str, ctx: Context) -> str:
    """Rebuild the search index for a collection."""
    chunks = await plan_chunks(collection)

    for i, chunk in enumerate(chunks, start=1):
        await process(chunk)
        await ctx.report_progress(
            progress=i,
            total=len(chunks),
            message=f"Indexed {i} of {len(chunks)} chunks",
        )

    return f"Reindexed {collection}: {len(chunks)} chunks"

There is no notification decorator and no generic notify(method, params) call. You send notifications through the Context the SDK hands your handler, and the methods available are the ones the protocol defines — which is the point. A server cannot invent a notification method and expect a client to route it.

Resource Change Notifications

Notify clients when resources change:

@mcp.tool()
async def update_document(doc_id: str, content: str, ctx: Context) -> str:
    """Replace a document's contents."""
    await db.update(doc_id, content)
    await ctx.notify_resource_updated(f"doc://{doc_id}")
    return f"Updated doc://{doc_id}"

notify_resource_updated tells clients that one URI changed. Its siblings — notify_resources_changed, notify_tools_changed, notify_prompts_changed — say that the list changed, which is what you send when a document is created or deleted rather than edited. Sending the wrong one is a common cause of a client showing stale entries: it dutifully re-read a resource that still exists while never noticing the three that appeared beside it.

Subscribing to Updates

Subscribing is a request, not a notification — it has an id and it gets a response. That response is simply a stream that stays open:

# Client → server. Note the id: this is a request, so it expects a reply.
{
    "jsonrpc": "2.0",
    "id": 7,
    "method": "subscriptions/listen",
    "params": { ... },
    "_meta": {
        "io.modelcontextprotocol/protocolVersion": "2026-07-28",
        "io.modelcontextprotocol/clientCapabilities": { ... }
    }
}

# Server acknowledges, then streams notifications on that open response.
# Each one carries the subscription id so the client can correlate it:
#   _meta: { "io.modelcontextprotocol/subscriptionId": ... }

Because the stream's state is scoped to the request rather than the connection, a dropped channel is the client's problem to recover: it re-issues subscriptions/listen. Your server keeps nothing.

Handling Subscriptions

class SubscriptionManager:
    def __init__(self):
        self.subscriptions = {}

    def subscribe(self, client_id: str, types: list):
        self.subscriptions[client_id] = set(types)

    def should_notify(self, client_id: str, notification_type: str) -> bool:
        if client_id not in self.subscriptions:
            return True  # Default: receive all
        return notification_type in self.subscriptions[client_id]

subscriptions = SubscriptionManager()

async def notify_clients(notification_type: str, data: dict):
    for client_id in connected_clients:
        if subscriptions.should_notify(client_id, notification_type):
            await send_to_client(client_id, {
                "type": notification_type,
                "data": data
            })

Best Practices

• Keep notifications lightweight
• Include enough context to avoid follow-up requests
• Use appropriate notification types
• Handle disconnected clients gracefully

Now let's apply these patterns in a practical lab. :::