Implementation:Microsoft Autogen Studio MCP Callbacks

Metadata
Sources	python/packages/autogen-studio/autogenstudio/mcp/callbacks.py
Domains	MCP, Callbacks, WebSocket, AI_Sampling
Last Updated	2026-02-11 17:00 GMT

Overview

Description

The Studio MCP Callbacks module provides factory functions that create callback handlers for the Model Context Protocol (MCP) implementation in AutoGen Studio. It implements three primary callback handlers that bridge MCP protocol events to WebSocket communication: message handling, AI sampling requests, and user input elicitation.

The module acts as an event translation layer between the MCP protocol and the WebSocket-based UI, enabling real-time streaming of MCP activities, tool sampling requests, and interactive user prompts. All callbacks are designed to work asynchronously and integrate with the MCPWebSocketBridge for bidirectional communication.

Usage

This module is used internally by AutoGen Studio's MCP integration to create protocol callbacks:

• create_message_handler() - Streams MCP protocol messages and errors to the UI
• create_sampling_callback() - Handles AI sampling requests from MCP tools (currently returns default responses)
• create_elicitation_callback() - Manages interactive user input requests from tools with timeout handling

Code Reference

Source Location

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

Primary Functions

create_message_handler

def create_message_handler(bridge: MCPWebSocketBridge):
    """Create a message handler callback that streams MCP protocol messages to the UI"""
    async def message_handler(
        message: RequestResponder[ServerRequest, ClientResult] | ServerNotification | Exception,
    ) -> None:
        # Implementation handles protocol messages, errors, and notifications

create_sampling_callback

def create_sampling_callback(bridge: MCPWebSocketBridge):
    """Create a sampling callback that handles AI sampling requests from tools"""
    async def sampling_callback(
        context: RequestContext[Any, Any, Any],
        params: CreateMessageRequestParams,
    ) -> CreateMessageResult | ErrorData:
        # Implementation returns placeholder response

create_elicitation_callback

def create_elicitation_callback(
    bridge: MCPWebSocketBridge,
) -> Tuple[Any, Dict[str, asyncio.Future[ElicitResult | ErrorData]]]:
    """Create an elicitation callback that handles user input requests from tools"""
    async def elicitation_callback(
        context: RequestContext[Any, Any, Any],
        params: ElicitRequestParams,
    ) -> ElicitResult | ErrorData:
        # Implementation manages async user input with timeout

Import Statement

from autogenstudio.mcp.callbacks import (
    create_message_handler,
    create_sampling_callback,
    create_elicitation_callback
)

I/O Contract

create_message_handler

Parameter	Type	Description
bridge	MCPWebSocketBridge	WebSocket bridge instance for sending messages to UI

Returns: Async callable that accepts ServerRequest, ServerNotification, or Exception messages

create_sampling_callback

Parameter	Type	Description
bridge	MCPWebSocketBridge	WebSocket bridge instance for sending activity updates

Returns: Async callable that accepts RequestContext and CreateMessageRequestParams, returns CreateMessageResult or ErrorData

create_elicitation_callback

Parameter	Type	Description
bridge	MCPWebSocketBridge	WebSocket bridge instance for managing elicitation requests

Returns: Tuple of (async callable, pending elicitations dict)

Activity Messages Sent

Activity Type	Description
"protocol"	MCP protocol messages with method and parameters
"error"	Protocol or callback errors
"sampling"	AI sampling requests and responses
"elicitation"	User input requests and responses

Usage Examples

Setting Up Message Handler

from autogenstudio.mcp.callbacks import create_message_handler
from autogenstudio.mcp.wsbridge import MCPWebSocketBridge

# Initialize bridge with WebSocket connection
bridge = MCPWebSocketBridge(websocket, session_id)

# Create message handler
message_handler = create_message_handler(bridge)

# Use with MCP session
async with stdio_client(server_params) as (read, write):
    async with ClientSession(read, write, message_handler=message_handler) as session:
        # MCP messages will now be streamed to UI
        await session.initialize()

Handling Sampling Requests

from autogenstudio.mcp.callbacks import create_sampling_callback

# Create sampling callback
sampling_callback = create_sampling_callback(bridge)

# Register with MCP session
async with ClientSession(read, write, sampling_callback=sampling_callback) as session:
    # Tools requesting AI sampling will receive default response
    result = await session.call_tool("some_tool", {})

Managing User Elicitation

from autogenstudio.mcp.callbacks import create_elicitation_callback

# Create elicitation callback and get pending dict
elicitation_callback, pending_elicitations = create_elicitation_callback(bridge)

# Register with MCP session
async with ClientSession(read, write, elicitation_callback=elicitation_callback) as session:
    # When tool requests user input:
    # 1. Request is sent to UI via bridge.on_elicitation_request()
    # 2. Future is stored in pending_elicitations with 60s timeout
    # 3. User response completes the future
    # 4. Result returned to tool
    pass

Handling Elicitation Responses

# In WebSocket message handler:
async def handle_elicitation_response(request_id: str, user_data: dict):
    if request_id in bridge.pending_elicitations:
        future = bridge.pending_elicitations[request_id]
        result = ElicitResult(action="accept", content=user_data)
        future.set_result(result)

Implementation Details

Message Handler Processing

The message handler categorizes incoming MCP messages:

• Exception - Extracts error details and sends as "error" activity
• ServerRequest with method - Sends protocol activity with method name and serialized params
• Other message types - Serializes message with type name

Sampling Response Behavior

The sampling callback currently returns a placeholder response:

CreateMessageResult(
    role="assistant",
    content=TextContent(
        type="text",
        text="[AutoGen Studio Default Sampling Response...]"
    ),
    model="autogen-studio-default"
)

This is intended for production configuration with a real LLM.

Elicitation Timeout

User elicitation requests have a 60-second timeout:

• Request sent to UI with unique request_id
• asyncio.Future stored in pending_elicitations dict
• await asyncio.wait_for(future, timeout=60.0)
• On timeout: ErrorData(-32603, "User did not respond...")
• Future cleanup happens in finally block

Error Handling

All callbacks implement comprehensive error handling:

• Extract "real" error messages via extract_real_error()
• Log errors via loguru logger
• Send error activities to UI via bridge
• Return ErrorData for protocol compliance

Related Pages

• Microsoft_Autogen_Studio_MCP_WSBridge - WebSocket bridge that receives callback events
• Microsoft_Autogen_Studio_MCP_Client - MCP client that uses these callbacks
• Implementation:MCP_Utils - Utility functions like extract_real_error() and serialize_for_json()
• Domain:MCP - Model Context Protocol domain
• Domain:WebSocket - WebSocket communication patterns