azure-ai-agentserver-responses samples for Python

Browse code

Quick start

pip install -r requirements.txt
python sample_01_getting_started.py

Samples index

# Sample Pattern Description
01 Getting Started TextResponse Echo handler — simplest async handler that echoes user input
02 Streaming Text Deltas TextResponse + text=iterable Token-by-token streaming via async iterable, with configure callback
03 Full Control ResponseEventStream Convenience, streaming, and builder — three ways to emit the same output
04 Function Calling ResponseEventStream Two-turn function calling with convenience and builder variants
05 Conversation History TextResponse + text=callable Study tutor with context.get_history() and ResponsesServerOptions
06 Multi-Output ResponseEventStream Math solver: reasoning + message, convenience and builder variants
07 Customization TextResponse Custom ResponsesServerOptions, default model, debug logging
08 Mixin Composition TextResponse Multi-protocol server via cooperative mixin inheritance
09 Self-Hosting TextResponse Mount responses into an existing Starlette app under /api
10 Streaming Upstream Raw events Forward to upstream streaming LLM via openai SDK, relay SSE events
11 Non-Streaming Upstream ResponseEventStream Forward to upstream non-streaming LLM via openai SDK, emit items
12 Image Generation ResponseEventStream Image gen convenience, streaming partials, and full-control builder
13 Image Input ResponseContext Receive images via URL, base64 data URL, or file ID
14 File Inputs ResponseContext Receive files via base64 data URL, URL, or file ID
15 Annotations ResponseEventStream Attach file_path, file_citation, and url_citation annotations to messages
16 Structured Outputs ResponseEventStream Return structured JSON as a structured_outputs item
19 Resilient Streaming Resilient Canonical framework-checkpoint handler with resilient_background=True — one output item per phase + stream.checkpoint(); on recovery seeds from context.persisted_response and resumes past the checkpointed phases (re-emitting the same items with their original ids)
20 Resilient Steering Resilient + steerable Demonstrates context.is_steered_turn / context.pending_input_count with resilient_background=True, steerable_conversations=True; naive re-run recovery (no recovery-specific code — the turn re-runs)
21 Resilient LangGraph Resilient + steerable Real-time streaming LangGraph agent (token-by-token deltas) composing an external durable engine: AsyncSqliteSaver (graph resume) + framework stream.checkpoint() / context.persisted_response (client items + ids). Checkpoints 1:1 with the graph, storing the graph checkpoint id in internal_metadata, so recovery rewinds to the point matching the persisted items — one reply, original ids, no divergence window
22 Resilient Multiturn Resilient Multi-turn conversation with resilient_background=True, steerable_conversations=False and explicit FoundryStateStore state

When to use which

  • TextResponse — Use for text-only responses (samples 1, 2, 5, 7–9). Handles the full SSE lifecycle automatically.
  • ResponseEventStream — Use when you need function calls, reasoning items, multiple output types, image generation, structured outputs, annotations, upstream proxying, or fine-grained event control (samples 3, 4, 6, 10–12, 15, 16).
  • ResponseContext — Use get_input_items() to inspect incoming images and files (samples 13, 14). Use context.is_recovery, context.is_steered_turn, and context.pending_input_count for resilient / steerable handlers; keep durable application state in FoundryStateStore (samples 19–22).

Enabling resilience and steering

Resilient + steerable behaviour is opt-in via ResponsesServerOptions — the defaults are both False. The resilient samples (19–22) each show the exact options shape they require; in short:

from azure.ai.agentserver.responses import ResponsesAgentServerHost, ResponsesServerOptions

app = ResponsesAgentServerHost(
    options=ResponsesServerOptions(
        resilient_background=True,             # opt-in to crash recovery
        steerable_conversations=True,        # opt-in to mid-turn steering
    ),
)

Either option initializes the Responses task runtime automatically; no set_resilient_tasks_enabled call is needed. That switch controls Invocations, not Responses.

With steerable_conversations=True and resilient_background=False, steering uses the normal persistent task backend. On restart, an interrupted stored response is marked failed without re-invoking its handler; persisted queued turns may still be promoted and executed.

With both options False, there is no Responses task runtime or task-based crash cleanup. An interrupted stored response can remain in_progress after a hard crash. Graceful shutdown still attempts to persist a failed terminal.

When the task runtime is active, concurrent requests for the same explicit conversation queue with steering enabled or receive 409 conversation_locked with steering disabled. Without a task runtime, overlapping requests execute independently rather than acquiring a task-backed conversation lock.