Client: Handling Streams and Progress
Relevant source files
The following files were used as context for generating this wiki page:The MCP Python SDK provides robust mechanisms for handling asynchronous, streaming communication between a client and a server. This is crucial for features like long-running tasks, real-time updates, and progress reporting. The ClientSession class is the primary interface for managing this communication, supported by various transport layers like WebSocket, SSE, and a custom StreamableHTTP protocol.
This document outlines how clients can receive server-sent notifications, handle progress updates for server-side tasks, and report progress for client-side operations. It covers the key components, data flows, and patterns for implementing these features effectively.
Core Components
ClientSession
The ClientSession is the central component for client-side interaction with an MCP server. It manages the connection, sends requests, and processes incoming messages, including responses and unsolicited notifications from the server.
Key responsibilities include:
- Initializing the session with the server.
- Sending requests like
call_tool,read_resource, etc. - Receiving and dispatching server messages through a
message_handler. - Providing callbacks for features like progress reporting (
progress_callback) and user interaction (elicitation_callback).
Sources: tests/server/fastmcp/test_integration.py:241-243, tests/shared/test_streamable_http.py:730-739
Notification Handling
Clients can receive asynchronous notifications from the server. These are handled by a message_handler callback provided during ClientSession instantiation. This handler receives all incoming messages, allowing the client to filter and process notifications like ProgressNotification, LoggingMessageNotification, or ResourceListChangedNotification.
The diagram below shows the flow for handling server-sent notifications.
This diagram illustrates how a message_handler is used to process notifications.
Sources: tests/server/fastmcp/test_integration.py:408-412, tests/shared/test_streamable_http.py:821-828
A common pattern for testing is to use a collector class to gather notifications.
class NotificationCollector:
"""Collects notifications from the server for testing."""
def __init__(self):
self.progress_notifications: list[ProgressNotificationParams] = []
# ... other notification lists
async def handle_generic_notification(
self, message: RequestResponder[ServerRequest, ClientResult] | ServerNotification | Exception
) -> None:
"""Handle any server notification and route to appropriate handler."""
if isinstance(message, ServerNotification):
if isinstance(message.root, ProgressNotification):
self.progress_notifications.append(message.root.params)
# ... other notification types
Sources: tests/server/fastmcp/test_integration.py:73-90
Progress Reporting
The SDK supports bidirectional progress reporting, allowing both the client and server to send updates during long-running operations.
Receiving Server Progress
When a client initiates a long-running task on the server (e.g., via call_tool), it can receive progress updates by providing a progress_callback. The server sends progress/notification messages, which the ClientSession directs to this callback.
The following sequence diagram shows a client calling a tool and receiving progress updates.
This diagram shows the flow of a tool call with progress notifications from the server.
Sources: tests/server/fastmcp/test_integration.py:431-443, tests/shared/test_progress_notifications.py:90-116
The ProgressNotification contains the following parameters:
| Parameter | Type | Description |
|---|---|---|
progressToken | str or int | A unique token to identify the progress stream. |
progress | float | The current progress value. |
total | float or None | The total value for the progress, if known. |
message | str or None | An optional message describing the current step. |
| Sources: tests/shared/test_progress_notifications.py:61-72 |
Sending Client Progress
The client can also report its own progress to the server. This is useful when the server delegates a task to the client. The SDK provides a progress context manager to simplify this process.
The context manager handles accumulating progress and sending notifications to the server.
The progress Context Manager
The progress context manager is the recommended way for clients to send progress updates. It is initialized with a RequestContext and an optional total value.
from mcp.shared.progress import progress
# ... inside an async function with access to a RequestContext
with progress(typed_context, total=100) as p:
await p.progress(10, message="Loading configuration...")
await p.progress(30, message="Connecting to database...")
await p.progress(40, message="Fetching data...")
await p.progress(20, message="Processing results...")
Sources: tests/shared/test_progress_notifications.py:261-266, src/mcp/shared/progress.py
The progress method of the context manager takes a delta (the amount of progress for the current step) and an optional message. It automatically calculates the cumulative progress and sends a progress/notification to the server.
The following diagram illustrates the internal logic of the progress context manager.
This diagram shows the flow of how the progress context manager sends updates.
Sources: src/mcp/shared/progress.py, tests/shared/test_progress_notifications.py:230-288
Transport-Specific Behavior
While the ClientSession API abstracts away the transport details, the underlying implementation of streaming and notifications varies.
StreamableHTTP
The StreamableHTTP transport uses a combination of POST and GET requests to handle bidirectional communication.
- POST Requests: Client sends requests (like
initializeortools/call) via POST. The response, including any immediate notifications, is streamed back as Server-Sent Events (SSE). - GET Stream: After initialization, the client makes a long-lived GET request. The server uses this connection to push asynchronous, unsolicited notifications (like
ResourceUpdatedNotificationorLoggingMessageNotification) to the client.
This dual-channel approach allows for request-response interactions on the POST channel while maintaining a persistent channel for server-initiated events on the GET channel.
Sources: tests/shared/test_streamable_http.py:818-842, tests/shared/test_streamable_http.py:650-681
WebSocket and SSE
In transports like WebSocket or standard SSE, a single persistent connection is used for all bidirectional communication, including requests, responses, and notifications. The ClientSession multiplexes messages over this single stream.
Sources: tests/shared/test_ws.py:63-66, tests/server/fastmcp/test_integration.py:202-205
Summary
The MCP Python SDK offers a powerful and flexible system for handling streaming data and progress notifications. By leveraging the ClientSession and its associated callbacks and context managers, developers can build responsive applications that provide real-time feedback for long-running tasks. The message_handler is key for processing asynchronous server notifications, while the progress_callback and progress context manager provide a clean, bidirectional API for progress reporting. The SDK effectively abstracts transport-level complexities, allowing developers to focus on application logic.