Azure AI Agent Server Responses client library for Python

The azure-ai-agentserver-responses package provides the Responses protocol endpoints for Azure AI Hosted Agent containers. It plugs into the azure-ai-agentserver-core host framework and adds the full response lifecycle: create, stream (SSE), cancel, delete, replay, and input-item listing.

Getting started

Install the package

pip install azure-ai-agentserver-responses

This automatically installs azure-ai-agentserver-core as a dependency.

Prerequisites

• Python 3.10 or later

Key concepts

ResponsesAgentServerHost

ResponsesAgentServerHost is an AgentServerHost subclass that adds Responses protocol endpoints. Register your handler with the @app.response_handler decorator:

@app.response_handler
async def my_handler(request: CreateResponse, context: ResponseContext, cancellation_signal: asyncio.Event):
    ...

Handlers MUST be async def and take exactly three positional parameters (request, context, cancellation_signal). Sync handlers and the 2-arg signature (request, context) are hard-rejected at decoration time. Cancellation is observed via the cancellation_signal event (set on client cancel, /cancel API, or steering pressure). Server shutdown is a distinct signal observed via context.shutdown — shutdown does NOT fire the cancellation signal; handlers that care about both must inspect each independently. See the handler implementation guide for the full surface.

Protocol endpoints

User isolation and response identifiers

The host scopes live execution, cancellation, SSE replay, and resilient task state by context.platform_context.user_id_key. Different users can run the same public response ID independently. Anonymous requests use a separate partition. The platform must supply trusted identity headers; accepting a caller-supplied user header without authentication is not an isolation boundary.

A duplicate POST /responses within the same user partition returns 409 response_id_conflict while execution or replay is retained. This does not prevent the owner from retrieving, reconnecting to, or cancelling the existing response; it is not an idempotent retry of POST. With file-backed replay, Core 2.3.0b1 or later also discovers retained logs after restart for admission, authorized replay, and deletion. Missing logs are not created by lookups, and expired logs no longer reserve the response ID. DELETE keeps the same user-scoped ID reserved through stream, runtime, and provider cleanup. New POSTs cannot reuse it midway through deletion; other users' partitions remain independent.

Storage providers must also enforce the supplied PlatformContext for response, item, and history operations. Runtime partitioning does not make a shared custom provider safe. Stored SSE replay requires a successful provider lookup for the requesting user before accessing that user's stream.

User-scoped internal identifiers do not change response IDs on the wire. Legacy shared task chains and replay logs are not used for identified users; drain active work before upgrading. Anonymous identifiers retain their previous format for recovery compatibility.

TextResponse

The simplest way to return text. Handles the full SSE lifecycle automatically (response.created → response.in_progress → message/content events → response.completed):

return TextResponse(context, request, text="Hello!")

For streaming, pass an async iterable to text:

async def tokens():
    for t in ["Hello", ", ", "world!"]:
        yield t

return TextResponse(context, request, text=tokens())

ResponseEventStream

Use ResponseEventStream when you need function calls, reasoning items, multiple output types, or fine-grained event control. Each yield maps 1:1 to an SSE event with zero bookkeeping:

stream = ResponseEventStream(response_id=context.response_id, request=request)
yield stream.emit_created()
yield stream.emit_in_progress()
yield from stream.output_item_message("Hello, world!")
yield stream.emit_completed()

Drop down to the builder API for full control over individual events:

message = stream.add_output_item_message()
yield message.emit_added()
text = message.add_text_content()
yield text.emit_added()
yield text.emit_delta("Hello!")
yield text.emit_text_done()
yield text.emit_done()
yield message.emit_done()

ResponseContext

The ResponseContext provides request-scoped state:

Persist cross-turn application state explicitly with azure.ai.agentserver.core.storage.FoundryStateStore, using conversation_chain_id as part of the store name.

The per-request cancellation signal is delivered as the 3rd positional handler argument (cancellation_signal: asyncio.Event), not via a ResponseContext attribute. It fires on client cancel (/cancel API or non-bg POST disconnect) or steering pressure; it does NOT fire on server shutdown — context.shutdown is the independent surface for that case.

Streaming and background modes

The SDK automatically handles all combinations of stream and background flags:

• Default — Run to completion, return final JSON response
• Streaming — Pipe events as SSE in real-time, cancel on client disconnect
• Background — Return immediately, handler runs in the background
• Streaming + Background — SSE while connected, handler continues after disconnect

Telemetry flushing

AGENTSERVER_FLUSH_MODE controls per-request telemetry flushing independently of the response's background flag:

Values are case-insensitive and whitespace is ignored. Unset, empty, or invalid values use background; each distinct invalid value logs a warning once. Streaming requests dispatch one flush after stream cleanup and before HTTP completion.

Background flushing requires the platform to allow the process to keep running long enough to drain telemetry before suspension or shutdown. If that window is unavailable, set AGENTSERVER_FLUSH_MODE=async to retain request-awaited flushing. Response completion alone does not guarantee telemetry delivery.

Response lifecycle

The library orchestrates the complete response lifecycle: created → in_progress → completed (or failed / cancelled). Cancellation, error handling, and terminal event guarantees are all managed automatically.

For detailed handler implementation guidance, see docs/handler-implementation-guide.md.

Resilience

Responses requests its task runtime at host startup when either ResponsesServerOptions(resilient_background=True) or steerable_conversations=True. This activation is protocol-scoped: Responses does not read or change Core's process-wide set_resilient_tasks_enabled switch. That switch remains the opt-in for direct Core tasks and Invocations, and cannot force-enable Responses when both Responses options are False.

With both options at their default False, stored responses run non-durably; there is no Responses recovery-time mark-failed or handler re-invocation. steerable_conversations=True alone enables the persistent steering queue/cancel backend: after a crash, the interrupted response is marked failed with code=server_error and its handler is not re-run, while persisted queued turns may be promoted as new steered turns. With resilient_background=True, stored background handlers are re-invoked after a crash; steering remains disabled unless its option is also enabled. Foreground stored responses are never re-invoked.

File-backed SSE replay, resilient-provider startup validation, response checkpoints, and shutdown preservation remain tied to resilient_background=True (with store=true, background=true where applicable); steering alone uses in-memory SSE replay. Request validation and connection behavior are unchanged: background=true, store=false is rejected, foreground disconnects cancel foreground work, and accepted background work survives client disconnect. For the complete request matrix and advanced recovery guidance, see the Resilient Responses Developer Guide.

Examples

Echo handler

from azure.ai.agentserver.responses import (
    CreateResponse,
    ResponseContext,
    ResponsesAgentServerHost,
    TextResponse,
)

app = ResponsesAgentServerHost()


@app.response_handler
async def handler(request: CreateResponse, context: ResponseContext, cancellation_signal: asyncio.Event):
    text = await context.get_input_text()
    return TextResponse(context, request, text=f"Echo: {text}")


app.run()

Multi-user session (per-request call ID)

On container protocol 2.0.0 a single agent session can serve multiple users. Forwarding the per-request x-agent-foundry-call-id on outbound toolbox calls lets the tool server resolve which user made this request and act on their behalf — so user A's and user B's requests to the same session each get a user-scoped result. (x-agent-user-id is never forwarded; the tool resolves the user from the call ID server-side. Use context.platform_context.user_id_key only for the container's own per-user state.)

import asyncio
import os

import httpx
from azure.ai.agentserver.core import get_request_context
from azure.ai.agentserver.responses import (
    CreateResponse,
    ResponseContext,
    ResponsesAgentServerHost,
    TextResponse,
)

app = ResponsesAgentServerHost()


@app.response_handler
async def handler(request: CreateResponse, context: ResponseContext, cancellation_signal: asyncio.Event):
    # platform_headers() echoes x-agent-foundry-call-id only (never x-agent-user-id).
    headers = get_request_context().platform_headers()

    # Toolbox / MCP — attach the call ID PER CALL. The MCP session is long-lived and
    # shared across users/turns, so never bake one call's ID into static client headers.
    async with httpx.AsyncClient() as mcp:
        resp = await mcp.post(
            f"{os.environ['FOUNDRY_PROJECT_ENDPOINT']}/toolboxes/github/mcp",
            headers={"Authorization": f"Bearer {get_agent_token()}", **headers},  # get_agent_token(): the agent's managed-identity token
            json={"jsonrpc": "2.0", "method": "tools/call",
                  "params": {"name": "list_my_assigned_issues", "arguments": {}}},
        )
        # The toolbox resolved the caller from the call ID and returned THIS user's issues.

    return TextResponse(context, request, text=resp.text)


app.run()

Function calling

import json

from azure.ai.agentserver.responses import ResponseEventStream

stream = ResponseEventStream(response_id=context.response_id, request=request)
yield stream.emit_created()
yield stream.emit_in_progress()

arguments = json.dumps({"location": "Seattle", "unit": "fahrenheit"})
yield from stream.output_item_function_call("get_weather", "call_001", arguments)

yield stream.emit_completed()

Reasoning + text message

stream = ResponseEventStream(response_id=context.response_id, request=request)
yield stream.emit_created()
yield stream.emit_in_progress()

yield from stream.output_item_reasoning_item("Let me think about this...")
yield from stream.output_item_message("Here is my answer.")

yield stream.emit_completed()

Configuration

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

options = ResponsesServerOptions(
    default_model="gpt-4o",
    sse_keep_alive_interval_seconds=15,
    shutdown_grace_period_seconds=10,
)

app = ResponsesAgentServerHost(options=options)

Troubleshooting

In-memory storage identity

InMemoryResponseProvider partitions response envelopes, input/output items, and history by the exact PlatformContext.user_id_key supplied by a trusted host. Omitting context or setting the user key to None selects a separate anonymous partition for local development, not unrestricted access to named users' data. Empty and whitespace keys remain distinct; call_id does not affect the partition. Pass the same user key on every related provider operation, including the legacy execution/replay helpers when used directly.

The provider does not authenticate callers. Hosts must establish trustworthy platform context before accessing it. This storage boundary is not complete end-to-end authorization: runtime routing and the process-wide SSE stream registry are separate from the provider's envelope, item, and history storage.

Common errors

• 400 Bad Request: The request body failed validation. Check that optional fields such as model (when provided) are valid and that input items are well-formed.
• 404 Not Found: The response ID does not exist. In hosted deployments persisted responses live in the Foundry hosted responses store; in local development they live under ${AGENTSERVER_STATE_ROOT:-~/.agentserver}/responses/ by default. A missing record may indicate the response was never persisted or was deleted via DELETE /responses/{id}.
• 400 Bad Request (cancel): The response was not created with background=true, or it has already reached a terminal state.

Reporting issues

To report an issue with the client library, or request additional features, please open a GitHub issue here.

Next steps

Visit the Samples folder for complete working examples:

• Handler implementation guide — Detailed reference for building handlers

Contributing

This project welcomes contributions and suggestions. Most contributions require you to agree to a Contributor License Agreement (CLA) declaring that you have the right to, and actually do, grant us the rights to use your contribution. For details, visit https://cla.microsoft.com.

When you submit a pull request, a CLA-bot will automatically determine whether you need to provide a CLA and decorate the PR appropriately (e.g., label, comment). Simply follow the instructions provided by the bot. You will only need to do this once across all repos using our CLA.

This project has adopted the Microsoft Open Source Code of Conduct. For more information, see the Code of Conduct FAQ or contact opencode@microsoft.com with any additional questions or comments.