Implementation:Microsoft Autogen Studio MCP Routes

Metadata	Value
Sources	Microsoft_Autogen
Domains	Model Context Protocol, WebSocket Communication, API Routes
Last Updated	2026-02-11 17:00 GMT

Overview

Description

The MCP Routes module provides FastAPI endpoints for establishing and managing Model Context Protocol (MCP) connections via WebSocket. It implements a bridge architecture that connects web clients to MCP servers through various transport mechanisms: stdio (command-line processes), SSE (Server-Sent Events), and StreamableHTTP. The module handles session lifecycle management, tracks active connections, provides status endpoints, and implements comprehensive error handling with proper WebSocket cleanup. It uses a callback system for message handling, sampling, and elicitation, enabling bidirectional communication between web clients and MCP servers.

Usage

The MCP Routes module is integrated into AutoGen Studio's FastAPI application as a router. Clients initiate connections by: 1. Calling POST /api/mcp/ws/connect with server parameters to get a WebSocket URL 2. Connecting to the returned WebSocket URL with encoded server parameters 3. Receiving initialization messages and server capabilities 4. Exchanging MCP protocol messages through the WebSocket bridge 5. Optionally checking session status via GET /api/mcp/ws/status/{session_id}

The module automatically handles connection lifecycle, including session tracking, error propagation, and cleanup when connections close.

Code Reference

Source Location

Repository: https://github.com/microsoft/autogen
File: python/packages/autogen-studio/autogenstudio/web/routes/mcp.py
Lines: 1-241

Signature

# FastAPI Router
router = APIRouter()

# Request Models
class CreateWebSocketConnectionRequest(BaseModel):
    server_params: McpServerParams

# Core Functions
async def create_mcp_session(
    bridge: MCPWebSocketBridge,
    server_params: McpServerParams,
    session_id: str
) -> None:
    """Create MCP session based on server parameters"""
    pass

# API Endpoints
@router.websocket("/ws/{session_id}")
async def mcp_websocket(websocket: WebSocket, session_id: str) -> None:
    """Main WebSocket endpoint - establishes MCP connection"""
    pass

@router.post("/ws/connect")
async def create_mcp_websocket_connection(
    request: CreateWebSocketConnectionRequest
) -> Dict[str, Any]:
    """Create WebSocket connection URL with encoded parameters"""
    pass

@router.get("/ws/status/{session_id}")
async def get_mcp_session_status(session_id: str) -> Dict[str, Any]:
    """Get MCP session status and capabilities"""
    pass

Import

from autogenstudio.web.routes.mcp import router
from fastapi import FastAPI

app = FastAPI()
app.include_router(router, prefix="/api/mcp")

I/O Contract

Inputs

POST /ws/connect

Parameter	Type	Description	Required
server_params	McpServerParams	Union type: StdioServerParams, SseServerParams, or StreamableHttpServerParams containing connection details	Yes

WebSocket /ws/{session_id}

Parameter	Type	Description	Required
session_id	str	Unique session identifier (path parameter)	Yes
server_params	str (query)	Base64-encoded JSON server parameters	Yes

GET /ws/status/{session_id}

Parameter	Type	Description	Required
session_id	str	Unique session identifier to query	Yes

Server Parameter Types:

• StdioServerParams: command (str), args (list), env (dict, optional)
• SseServerParams: url (str)
• StreamableHttpServerParams: url (str)

Outputs

POST /ws/connect Response

Field	Type	Description
status	bool	Success status of URL creation
message	str	Human-readable status message
session_id	str	Generated UUID for the session
websocket_url	str	WebSocket URL with encoded parameters
timestamp	str	ISO format timestamp of creation

GET /ws/status/{session_id} Response

Field	Type	Description
status	bool	Whether session was found
message	str	Status message
session_id	str	Queried session identifier
connected	bool	Whether session is currently active
capabilities	dict	MCP server capabilities if available
created_at	str	ISO format session creation timestamp
last_activity	str	ISO format last activity timestamp

WebSocket Messages (Bidirectional)

• Client to Server: MCP protocol messages (initialize, call_tool, list_resources, etc.)
• Server to Client: MCP responses, capabilities, tool results, errors, status updates

Usage Examples

Creating WebSocket Connection

import requests
import json

# Create connection for stdio server
connection_request = {
    "server_params": {
        "type": "StdioServerParams",
        "command": "python",
        "args": ["-m", "mcp_server"],
        "env": {"API_KEY": "secret"}
    }
}

response = requests.post(
    "http://localhost:8000/api/mcp/ws/connect",
    json=connection_request
)

result = response.json()
print(f"Session ID: {result['session_id']}")
print(f"WebSocket URL: {result['websocket_url']}")

Connecting via WebSocket Client

import asyncio
import websockets
import json

async def connect_to_mcp():
    # Get connection URL from /ws/connect endpoint first
    ws_url = "ws://localhost:8000/api/mcp/ws/abc-123?server_params=..."

    async with websockets.connect(ws_url) as websocket:
        # Receive initialization message
        init_msg = await websocket.recv()
        print(f"Initialized: {init_msg}")

        # Send list_tools request
        request = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "tools/list"
        }
        await websocket.send(json.dumps(request))

        # Receive response
        response = await websocket.recv()
        tools = json.loads(response)
        print(f"Available tools: {tools}")

asyncio.run(connect_to_mcp())

Checking Session Status

import requests

session_id = "550e8400-e29b-41d4-a716-446655440000"
response = requests.get(
    f"http://localhost:8000/api/mcp/ws/status/{session_id}"
)

status = response.json()
if status["status"]:
    print(f"Session active since: {status['created_at']}")
    print(f"Capabilities: {status['capabilities']}")
else:
    print("Session not found")

Creating SSE Server Connection

import requests

# Connect to SSE-based MCP server
connection_request = {
    "server_params": {
        "type": "SseServerParams",
        "url": "https://mcp-server.example.com/sse"
    }
}

response = requests.post(
    "http://localhost:8000/api/mcp/ws/connect",
    json=connection_request
)

result = response.json()
print(f"SSE WebSocket URL: {result['websocket_url']}")

Creating StreamableHTTP Connection

import requests

# Connect to StreamableHTTP MCP server
connection_request = {
    "server_params": {
        "type": "StreamableHttpServerParams",
        "url": "https://mcp-server.example.com/stream"
    }
}

response = requests.post(
    "http://localhost:8000/api/mcp/ws/connect",
    json=connection_request
)

result = response.json()
print(f"StreamableHTTP WebSocket URL: {result['websocket_url']}")

Full Client Integration Example

import asyncio
import requests
import websockets
import json

async def mcp_client_example():
    # Step 1: Create connection
    connection_request = {
        "server_params": {
            "type": "StdioServerParams",
            "command": "uvx",
            "args": ["mcp-server-sqlite"],
            "env": {}
        }
    }

    response = requests.post(
        "http://localhost:8000/api/mcp/ws/connect",
        json=connection_request
    )
    connection = response.json()
    session_id = connection["session_id"]
    ws_url = f"ws://localhost:8000{connection['websocket_url']}"

    # Step 2: Connect and interact
    async with websockets.connect(ws_url) as websocket:
        # Wait for initialization
        init = await websocket.recv()
        print(f"Connected: {init}")

        # List available tools
        await websocket.send(json.dumps({
            "jsonrpc": "2.0",
            "id": 1,
            "method": "tools/list"
        }))

        tools_response = await websocket.recv()
        print(f"Tools: {tools_response}")

        # Call a tool
        await websocket.send(json.dumps({
            "jsonrpc": "2.0",
            "id": 2,
            "method": "tools/call",
            "params": {
                "name": "query",
                "arguments": {"sql": "SELECT * FROM users LIMIT 10"}
            }
        }))

        result = await websocket.recv()
        print(f"Result: {result}")

    # Step 3: Check status
    status = requests.get(
        f"http://localhost:8000/api/mcp/ws/status/{session_id}"
    )
    print(f"Final status: {status.json()}")

asyncio.run(mcp_client_example())

Related Pages

• MCP Client - MCPClient class used internally
• MCP WebSocket Bridge - Bridge architecture for WebSocket communication
• MCP Callbacks - Callback functions for message handling
• Studio Web Server - Main FastAPI application
• Model Context Protocol - MCP specification and protocol details
• WebSocket Communication - WebSocket patterns in AutoGen