Real-Time Updates and Notifications

MCP supports server-initiated notifications, enabling real-time updates to connected clients.

Notification Types

Type	Use Case
Resource updates	Data changed
Progress updates	Long-running tasks
System alerts	Errors, warnings

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

MCP hosts can subscribe to specific notifications:

# Client subscription (host-side)
{
    "jsonrpc": "2.0",
    "method": "notifications/subscribe",
    "params": {
        "types": ["resources/updated", "progress"]
    }
}

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