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

@server.notification()
async def send_notification(method: str, params: dict):
    # MCP automatically routes to connected clients
    pass

# In your tool or background task
async def long_running_task():
    for i, chunk in enumerate(process_data()):
        # Send progress update
        await server.notify(
            "notifications/progress",
            {"task_id": "abc", "progress": i / 100}
        )

    # Send completion
    await server.notify(
        "notifications/complete",
        {"task_id": "abc", "result": "success"}
    )

Resource Change Notifications

Notify clients when resources change:

async def update_document(doc_id: str, content: str):
    # Update the document
    await db.update(doc_id, content)

    # Notify clients about the change
    await server.notify(
        "notifications/resources/updated",
        {"uri": f"doc://{doc_id}"}
    )

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. :::