Support for MCP Progress, Notifications, and Logging in MCP Toolset #3398
Description
📌 Description
Currently, the MCPToolset in the ADK-Python SDK (located at src/google/adk/tools/mcp_tool/mcp_toolset.py) enables interaction with external MCP (Model Control Protocol) tools, but it does not provide any mechanism to receive real-time progress updates, notifications, or logs emitted by those tools.
Many modern MCP tools (e.g., FastMCP) emit structured progress events and logging output over standardized channels (e.g., HTTP endpoints or WebSocket streams) to improve user experience and debugging. For example:
Progress: https://gofastmcp.com/servers/progress — allows tools to report task completion %, estimated time, or step-wise updates.
Logging: https://gofastmcp.com/servers/logging#client-handling — defines a client-side interface for receiving and displaying structured log lines (e.g., INFO, WARN, ERROR).
Without this support, ADK users are left with a "black box" experience — they initiate a tool but receive no feedback until completion (or failure), which reduces usability, debuggability, and trust in long-running operations.
✨ Requested Feature
Extend the MCPToolset class to optionally:
Subscribe to progress updates from the MCP server via the /progress endpoint.
Subscribe to real-time logs via the /logging endpoint (as described in the Client Handling spec).
Expose these events through callback functions or a pub/sub interface within the ADK Python environment.
This could be implemented as:
class MCPToolset:
def __init__(
self,
tool_definition: dict,
on_progress: Optional[Callable[[dict], None]] = None, # e.g., {"percent": 75, "step": "processing"}
on_log: Optional[Callable[[dict], None]] = None, # e.g., {"level": "INFO", "message": "Starting download"}
on_notification: Optional[Callable[[dict], None]] = None, # e.g., {"title": "Task complete", "type": "success"}
):
...Or via a more event-driven pattern using an EventStream listener that connects to the MCP server’s endpoints automatically when enabled.
🎯 Benefits
Improved UX: Users see real-time feedback during long-running tool invocations.
Better Debugging: Logs from the MCP tool appear in the ADK console or UI, reducing need for external log inspection.
Standardization: Aligns ADK-Python with industry-standard MCP behavior (FastMCP, etc.).
Future-proofing: Enables integration with advanced UIs (e.g., Android IDEs) that render progress bars or log streams.
🔗 References
FastMCP Progress Specification
FastMCP Logging Specification — Client Handling
Current MCPToolset Implementation
💡 Implementation Suggestion
Use requests or aiohttp (if async support is desired) to poll or listen to the /progress and /logging endpoints.
If the MCP tool provides a WebSocket or SSE (Server-Sent Events) endpoint, support that for true real-time streaming.
Add a new config flag in tool definition (e.g., enable_progress_logging: true) to opt-in to this behavior.
Log events can be forwarded to Python's logging module for compatibility.
Progress events can be surfaced to UI layers via callbacks or ADK’s internal event bus.
📌 Example Usage
def handle_progress(progress):
print(f"Progress: {progress['percent']}% - {progress.get('step', 'running')}")
def handle_log(log_entry):
if log_entry["level"] == "ERROR":
print(f"❌ {log_entry['message']}")
else:
print(f"ℹ️ {log_entry['message']}")
toolset = MCPToolset(
tool_definition=my_tool_def,
on_progress=handle_progress,
on_log=handle_log,
)
toolset.execute()✅ Acceptance Criteria
- MCPToolset can optionally receive progress updates from an MCP server.
- MCPToolset can optionally receive and display structured log entries.
- Documentation updated with examples and configuration options.
- Tests added for progress and log handling (mocked HTTP endpoints).
- No breaking changes to existing API unless necessary.