Implementation:Microsoft Autogen Studio MCP Client

Metadata
Sources	python/packages/autogen-studio/autogenstudio/mcp/client.py
Domains	MCP, Client, Protocol, Tools
Last Updated	2026-02-11 17:00 GMT

Overview

Description

The Studio MCP Client module provides a transport-agnostic Model Context Protocol (MCP) client implementation for AutoGen Studio. It defines the MCPClient class that handles core MCP protocol operations independently of the underlying transport mechanism (WebSocket, stdio, etc.).

The module implements the Protocol pattern with the MCPEventHandler interface, allowing flexible event handling while maintaining clean separation between protocol logic and event notification. The client supports all standard MCP operations: listing and calling tools, managing resources, and working with prompts.

Usage

This module is used to:

• Initialize MCP sessions with capability negotiation
• Execute MCP operations (list_tools, call_tool, list_resources, read_resource, list_prompts, get_prompt)
• Notify event handlers of operation results and errors
• Abstract transport details from protocol logic

The MCPClient works in conjunction with the MCP ClientSession and an event handler implementation.

Code Reference

Source Location

Repository: https://github.com/microsoft/autogen
File Path: python/packages/autogen-studio/autogenstudio/mcp/client.py
Lines: 121

Class Signature

class MCPEventHandler(Protocol):
    """Interface for handling MCP events"""
    async def on_initialized(self, session_id: str, capabilities: Any) -> None: ...
    async def on_operation_result(self, operation: str, data: Dict[str, Any]) -> None: ...
    async def on_operation_error(self, operation: str, error: str) -> None: ...
    async def on_mcp_activity(self, activity_type: str, message: str, details: Dict[str, Any]) -> None: ...
    async def on_elicitation_request(self, request_id: str, message: str, requested_schema: Any) -> None: ...

class MCPClient:
    """Handles MCP protocol operations independently of transport"""
    def __init__(self, session: ClientSession, session_id: str, event_handler: MCPEventHandler): ...
    async def initialize(self) -> None: ...
    async def handle_operation(self, operation: Dict[str, Any]) -> None: ...
    @property
    def capabilities(self): ...

Import Statement

from autogenstudio.mcp.client import MCPClient, MCPEventHandler

I/O Contract

Constructor Parameters

Parameter	Type	Description
session	ClientSession	MCP ClientSession instance for protocol communication
session_id	str	Unique identifier for the session
event_handler	MCPEventHandler	Handler implementing MCPEventHandler protocol for event callbacks

initialize()

Returns	Description
None	Initializes MCP session and notifies handler with capabilities

Raises: Exception on initialization failure (after notifying error handler)

handle_operation()

Parameter	Type	Description
operation	Dict[str, Any]	Operation dictionary with 'operation' key and operation-specific parameters

Supported Operations:

Operation	Required Parameters	Description
list_tools	-	Lists available MCP tools
call_tool	tool_name, arguments	Calls a specific tool with arguments
list_resources	-	Lists available resources
read_resource	uri	Reads a specific resource by URI
list_prompts	-	Lists available prompts
get_prompt	name, arguments (optional)	Retrieves a specific prompt template

Returns: None (results sent via event handler callbacks)

MCPEventHandler Protocol

Method	Parameters	Description
on_initialized	session_id: str, capabilities: Any	Called when session initializes successfully
on_operation_result	operation: str, data: Dict[str, Any]	Called when operation completes with results
on_operation_error	operation: str, error: str	Called when operation fails
on_mcp_activity	activity_type: str, message: str, details: Dict[str, Any]	Called for general MCP protocol activity
on_elicitation_request	request_id: str, message: str, requested_schema: Any	Called when tool requests user input

Usage Examples

Basic Client Initialization

from mcp import ClientSession
from autogenstudio.mcp.client import MCPClient, MCPEventHandler
from autogenstudio.mcp.wsbridge import MCPWebSocketBridge

# Create event handler (bridge implements MCPEventHandler)
bridge = MCPWebSocketBridge(websocket, session_id)

# Create MCP client
async with ClientSession(read, write) as session:
    client = MCPClient(session, session_id, bridge)

    # Initialize session
    await client.initialize()

    # Client is ready for operations
    print(f"Capabilities: {client.capabilities}")

Listing Available Tools

# Operation dictionary for listing tools
operation = {
    "operation": "list_tools"
}

# Handle the operation
await client.handle_operation(operation)

# Result will be sent to event_handler.on_operation_result()
# with data like: {"tools": [{"name": "tool1", ...}, ...]}

Calling a Tool

# Operation to call a tool
operation = {
    "operation": "call_tool",
    "tool_name": "fetch_weather",
    "arguments": {
        "location": "San Francisco",
        "units": "celsius"
    }
}

await client.handle_operation(operation)

# Result sent to event_handler.on_operation_result()
# with data like: {"tool_name": "fetch_weather", "result": {...}}

Reading a Resource

# Operation to read a resource
operation = {
    "operation": "read_resource",
    "uri": "file:///path/to/resource.txt"
}

await client.handle_operation(operation)

# Result sent to event_handler.on_operation_result()

Getting a Prompt Template

# Operation to get a prompt
operation = {
    "operation": "get_prompt",
    "name": "summarize_document",
    "arguments": {
        "document_type": "technical"
    }
}

await client.handle_operation(operation)

# Prompt template sent to event_handler.on_operation_result()

Custom Event Handler

from typing import Any, Dict

class CustomEventHandler:
    """Custom implementation of MCPEventHandler"""

    async def on_initialized(self, session_id: str, capabilities: Any) -> None:
        print(f"Session {session_id} initialized with capabilities: {capabilities}")

    async def on_operation_result(self, operation: str, data: Dict[str, Any]) -> None:
        print(f"Operation {operation} completed: {data}")

    async def on_operation_error(self, operation: str, error: str) -> None:
        print(f"Operation {operation} failed: {error}")

    async def on_mcp_activity(self, activity_type: str, message: str, details: Dict[str, Any]) -> None:
        print(f"Activity [{activity_type}]: {message}")

    async def on_elicitation_request(self, request_id: str, message: str, requested_schema: Any) -> None:
        print(f"Elicitation {request_id}: {message}")

# Use custom handler
handler = CustomEventHandler()
client = MCPClient(session, "custom-session", handler)

Implementation Details

Initialization Process

The initialize() method:

1. Calls session.initialize() to negotiate capabilities
2. Stores capabilities in _capabilities attribute
3. Sets _initialized flag to True
4. Serializes capabilities via serialize_for_json()
5. Notifies event handler via on_initialized()
6. On error: notifies handler and re-raises exception

Operation Dispatch

The handle_operation() method implements a dispatch pattern:

1. Extracts operation type from operation dict
2. Calls appropriate session method (list_tools, call_tool, etc.)
3. Serializes results using serialize_for_json()
4. Notifies handler via on_operation_result()
5. On error: extracts real error and notifies via on_operation_error()

Error Handling

Errors are processed through:

• extract_real_error() - Extracts meaningful error messages
• McpOperationError - Custom exception for operation-specific errors
• Event handler notification for all errors
• No silent failures - all errors propagated to handler

Serialization

All data sent to event handlers is serialized via serialize_for_json() to ensure:

• Pydantic models are converted to dicts
• Complex objects are JSON-compatible
• Timestamps and special types are handled

Related Pages

• Microsoft_Autogen_Studio_MCP_WSBridge - WebSocket bridge implementing MCPEventHandler
• Microsoft_Autogen_Studio_MCP_Callbacks - Callback factories for MCP events
• Implementation:MCP_Utils - Utility functions for serialization and error handling
• Domain:MCP - Model Context Protocol domain
• Domain:Protocol_Design - Protocol and interface patterns