Implementation:Microsoft Autogen Studio Auth Middleware

Metadata
Sources	python/packages/autogen-studio/autogenstudio/web/auth/middleware.py
Domains	Authentication, Middleware, FastAPI, WebSocket
Last Updated	2026-02-11 17:00 GMT

Overview

Description

The Studio Auth Middleware module implements two middleware classes for authentication in AutoGen Studio: AuthMiddleware for HTTP requests and WebSocketAuthMiddleware for WebSocket connections. The middleware intercepts all incoming requests, validates authentication tokens, and populates the request context with user information.

The HTTP middleware extends Starlette's BaseHTTPMiddleware and provides sophisticated path-based exclusion rules for static assets, public endpoints, and authentication routes. The WebSocket middleware provides token validation specifically for WebSocket connections, which require different handling than standard HTTP requests.

Usage

This module is used to:

• Authenticate HTTP requests before they reach route handlers
• Authenticate WebSocket connections before accepting them
• Exclude specific paths from authentication (static files, public endpoints)
• Inject user context into request.state for route handlers
• Return 401 responses for unauthorized requests

Code Reference

Source Location

Repository: https://github.com/microsoft/autogen
File Path: python/packages/autogen-studio/autogenstudio/web/auth/middleware.py
Lines: 120

Class Signatures

from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint

class AuthMiddleware(BaseHTTPMiddleware):
    """
    Middleware for handling authentication for all routes.
    """

    def __init__(self, app: ASGIApp, auth_manager: AuthManager) -> None: ...
    async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response: ...

class WebSocketAuthMiddleware:
    """
    Helper for authenticating WebSocket connections.
    Not a middleware in the traditional sense - used in WebSocket endpoint.
    """

    def __init__(self, auth_manager: AuthManager) -> None: ...
    async def authenticate(self, websocket: WebSocket) -> bool: ...

Import Statement

from autogenstudio.web.auth.middleware import AuthMiddleware, WebSocketAuthMiddleware

I/O Contract

AuthMiddleware

Parameter	Type	Description
app	ASGIApp	ASGI application instance
auth_manager	AuthManager	AuthManager instance for authentication

dispatch() Method

Parameter	Type	Description
request	Request	Incoming HTTP request
call_next	RequestResponseEndpoint	Next middleware or route handler

Returns: Response object

Side Effects: Sets request.state.user if authentication succeeds

WebSocketAuthMiddleware

Parameter	Type	Description
auth_manager	AuthManager	AuthManager instance for token validation

authenticate() Method

Parameter	Type	Description
websocket	WebSocket	WebSocket connection to authenticate

Returns: Boolean (True if authenticated, False otherwise)

Token Sources: Query parameter "token" or Authorization header

Usage Examples

Adding HTTP Middleware

from fastapi import FastAPI
from autogenstudio.web.auth.manager import AuthManager
from autogenstudio.web.auth.middleware import AuthMiddleware

app = FastAPI()

# Initialize auth manager
auth_manager = AuthManager.from_env()

# Add authentication middleware
app.add_middleware(AuthMiddleware, auth_manager=auth_manager)

@app.get("/api/protected")
async def protected_route(request: Request):
    # User automatically available in request.state
    user = request.state.user
    return {"user_id": user.id, "name": user.name}

WebSocket Authentication

from fastapi import WebSocket, WebSocketDisconnect
from autogenstudio.web.auth.middleware import WebSocketAuthMiddleware

# Create WebSocket auth middleware
ws_auth = WebSocketAuthMiddleware(auth_manager)

@app.websocket("/api/ws")
async def websocket_endpoint(websocket: WebSocket):
    # Authenticate before accepting connection
    if not await ws_auth.authenticate(websocket):
        await websocket.close(code=1008, reason="Unauthorized")
        return

    # Accept connection if authenticated
    await websocket.accept()

    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Echo: {data}")
    except WebSocketDisconnect:
        pass

WebSocket with Token in Query String

// Frontend JavaScript
const token = localStorage.getItem('auth_token');
const ws = new WebSocket(`ws://localhost:8000/api/ws?token=${token}`);

ws.onopen = () => {
    console.log('Connected');
    ws.send('Hello server');
};

WebSocket with Token in Header

# Python client
import asyncio
import websockets

token = "your-jwt-token"
headers = {"Authorization": f"Bearer {token}"}

async with websockets.connect(
    "ws://localhost:8000/api/ws",
    extra_headers=headers
) as websocket:
    await websocket.send("Hello server")
    response = await websocket.recv()
    print(response)

Complete Application Setup

from fastapi import FastAPI, Request, WebSocket
from autogenstudio.web.auth.manager import AuthManager
from autogenstudio.web.auth.middleware import AuthMiddleware, WebSocketAuthMiddleware
from autogenstudio.web.auth.authroutes import router as auth_router

app = FastAPI()

# Initialize authentication
auth_manager = AuthManager.from_env()
app.state.auth_manager = auth_manager

# Add HTTP middleware
app.add_middleware(AuthMiddleware, auth_manager=auth_manager)

# Create WebSocket middleware
ws_auth = WebSocketAuthMiddleware(auth_manager)

# Include auth routes
app.include_router(auth_router, prefix="/api/auth", tags=["auth"])

# Public endpoint (excluded in middleware)
@app.get("/api/health")
async def health_check():
    return {"status": "healthy"}

# Protected endpoint
@app.get("/api/user/profile")
async def get_profile(request: Request):
    user = request.state.user
    return {"id": user.id, "name": user.name}

# Protected WebSocket
@app.websocket("/api/ws/chat")
async def websocket_chat(websocket: WebSocket):
    if not await ws_auth.authenticate(websocket):
        await websocket.close(code=1008)
        return

    await websocket.accept()
    # Handle WebSocket messages...

Accessing User in Route

from fastapi import Request

@app.get("/api/current-user")
async def get_current_user(request: Request):
    # User set by middleware
    user = request.state.user

    return {
        "id": user.id,
        "name": user.name,
        "email": user.email,
        "roles": user.roles
    }

Implementation Details

Excluded Paths (HTTP Middleware)

The following paths are excluded from authentication:

Exact Matches:

• / (root - serves frontend)
• /login
• /callback
• /images
• Paths in auth_manager.config.exclude_paths

Pattern Matches:

• /page-data/* (Gatsby build artifacts)
• *.js, *.css, *.png, *.ico, *.svg, *.jpg, *.webmanifest, *.json (static assets)
• *.js.map, *.svg (source maps and SVGs)

WebSocket Paths:

• /api/ws/*
• /api/maker/*

if (
    path == "/"
    or path == "/login"
    or path == "/callback"
    or path == "/images"
    or path.startswith("/page-data/")
    or path in self.auth_manager.config.exclude_paths
    or re.match(r"/[^/]+\.(js|css|png|ico|svg|jpg|webmanifest|json)$", path)
    or re.match(r".*\.(js\.map|svg)$", path)
):
    return await call_next(request)

OPTIONS Request Handling

CORS preflight requests (OPTIONS) always bypass authentication:

if request.method == "OPTIONS":
    return await call_next(request)

WebSocket Path Handling

WebSocket endpoints are passed through in the HTTP middleware:

if request.url.path.startswith("/api/ws") or request.url.path.startswith("/api/maker"):
    # Auth handled in WebSocket accept handler
    return await call_next(request)

No-Auth Mode

When auth type is "none":

if self.auth_manager.config.type == "none":
    request.state.user = await self.auth_manager.authenticate_request(request)
    return await call_next(request)

User is still set in request.state for consistency.

Error Response Format

Unauthorized requests receive 401 with JSON response:

{
    "status": false,
    "detail": "Missing authentication token"
}

Headers from AuthException are included if present.

WebSocket Token Extraction

WebSocketAuthMiddleware checks two sources:

1. Query parameter: ?token=xxx
2. Authorization header: Authorization: Bearer xxx

token = None
if "token" in websocket.query_params:
    token = websocket.query_params["token"]
elif "authorization" in websocket.headers:
    auth_header = websocket.headers["authorization"]
    if auth_header.startswith("Bearer "):
        token = auth_header.replace("Bearer ", "")

WebSocket Validation

Token validation uses auth_manager.is_valid_token():

• Returns True if token is valid and not expired
• Returns False if token is invalid, expired, or missing
• Logs warnings for missing or invalid tokens
• No exception raising - just returns boolean

Exception Handling

The HTTP middleware catches three types of errors:

1. AuthException - Expected auth errors (401)
2. Unexpected Exception - Logged and returned as 401
3. All exceptions include JSON response body

Logging

The middleware uses loguru logger:

• Warnings for expired/invalid tokens
• Warnings for missing user in request.state
• Errors for unexpected exceptions
• Info level for token validation results

Related Pages

• Microsoft_Autogen_Studio_Auth_Manager - AuthManager used by middleware
• Microsoft_Autogen_Studio_Auth_Routes - Auth routes excluded from middleware
• Microsoft_Autogen_Studio_Auth_Models - User model set in request.state
• Implementation:Auth_Exceptions - AuthException and subclasses
• Domain:Middleware - Middleware patterns and best practices
• Domain:Authentication - Authentication concepts
• Domain:WebSocket - WebSocket protocol and authentication
• Domain:FastAPI - FastAPI framework features