Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs.xpander.ai/llms.txt

Use this file to discover all available pages before exploring further.

@on_auth_event registers a handler that fires whenever the agent runtime emits an authentication event: for example, when an MCP server requires the end-user to log in via OAuth. Use it to surface the OAuth URL to your UI, kick off external auth flows, or log auth attempts. 
from xpander_sdk import on_auth_event
from xpander_sdk.modules.agents.sub_modules.agent import Agent
from xpander_sdk.modules.tasks.sub_modules.task import Task, TaskUpdateEvent

@on_auth_event
async def handle_auth(agent: Agent, task: Task, event: TaskUpdateEvent):
    print(f"Auth required for {agent.name}")
    print(f"Task: {task.id}")
    print(f"Auth data: {event.data}")
The handler is auto-registered globally: you don’t pass it anywhere. Every Backend.aget_args(...) call routes auth events through registered handlers automatically.

​
Required signature

The function must accept exactly three parameters: agent, task, event. Names are flexible; arity matters. Sync or async are both accepted. 
@on_auth_event
async def handler(agent, task, event):     # ✓
    ...

@on_auth_event
def handler(agent, task, event):           # ✓ (sync)
    ...

@on_auth_event
def handler(agent, task):                  # ✗ TypeError: needs 3 params
    ...
ParameterTypeDescription
agentAgentThe agent processing the task.
taskTaskThe current task.
eventTaskUpdateEventevent.type == "auth_event". event.data holds the auth payload.
event.type is always "auth_event" for handlers registered with this decorator.

​
Event payload shapes

event.data is one of the MCP OAuth response variants. Switch on event.data.type:
typeevent.data.data
MCPOAuthResponseType.LOGIN_REQUIRED ("login_required")MCPOAuthGetTokenLoginRequiredResponse(url, server_url, server_name)
MCPOAuthResponseType.TOKEN_READY ("token_ready")MCPOAuthGetTokenTokenReadyResponse(access_token)
MCPOAuthResponseType.TOKEN_ISSUE ("token_issue")MCPOAuthGetTokenGenericResponse(message)
MCPOAuthResponseType.NOT_SUPPORTED ("not_supported")MCPOAuthGetTokenGenericResponse(message)

​
Examples

​
Show the OAuth URL to the user

@on_auth_event
async def show_oauth_url(agent, task, event):
    payload = event.data
    if payload.type == "login_required":
        login = payload.data
        await my_ui.show(f"Please authenticate with {login.server_name}: {login.url}")

​
Multiple handlers stack

@on_auth_event
async def log_auth(agent, task, event):
    audit_log.info(f"auth event for {agent.id}: {event.data}")

@on_auth_event
async def notify_user(agent, task, event):
    await pubsub.publish(f"user:{task.input.user.id}", event.data)
Both handlers fire on every auth event. Order is registration order.

​
Combine with a per-call callback

You can pass an additional one-off callback to Backend.aget_args(auth_events_callback=...): it runs in addition to every globally-registered handler. 
@on_auth_event
async def global_handler(agent, task, event):
    audit(event)

# Per-call extra:
async def per_call(agent, task, event):
    await my_ui.show_special_state(event)

args = await backend.aget_args(
    agent_id="agent-123",
    auth_events_callback=per_call,
)
Both global_handler and per_call fire when an auth event happens during the resulting agent’s run.

​
Helper functions

The module also exposes two helpers for testing and management: 
from xpander_sdk.modules.backend.decorators.on_auth_event import (
    get_registered_handlers,
    clear_handlers,
)

# Inspect what's registered
handlers = get_registered_handlers()

# Wipe registered handlers (useful between tests)
clear_handlers()
These aren’t re-exported from the top-level xpander_sdk package: import from the decorator module directly.