core Package

Azure AI AgentServerHost core framework.

Provides the <xref:AgentServerHost> base class and shared utilities for building Azure AI Hosted Agent containers.

Public API::

from azure.ai.agentserver.core import (
    AgentConfig,
    AgentServerHost,
    configure_observability,
    create_error_response,
    detach_context,
    end_span,
    flush_spans,
    record_error,
    set_current_span,
    trace_stream,
)

Classes

AgentConfig

Resolved configuration for an agent server host.

All values are populated from environment variables at creation time. Access via app.config:


   app = InvocationAgentServerHost()
   print(app.config.agent_name)
   print(app.config.project_endpoint)
AgentServerHost

Agent server host framework for Azure AI Hosted Agent containers.

A <xref:starlette.applications.Starlette> subclass providing the protocol-agnostic infrastructure required by all hosted agent containers:

  • Health probe (GET /readiness)

  • Graceful shutdown handling (SIGTERM, configurable timeout)

  • OpenTelemetry tracing with Azure Monitor and OTLP exporters

  • Hypercorn-based ASGI server with HTTP/1.1

Protocol packages subclass this host to add their own routes:


   from azure.ai.agentserver.invocations import InvocationAgentServerHost

   app = InvocationAgentServerHost()

   @app.invoke_handler
   async def handle(request):
       return JSONResponse({"ok": True})

   app.run()

For multi-protocol agents, compose via cooperative inheritance:


   class MyHost(InvocationAgentServerHost, ResponsesAgentServerHost):
       pass

   app = MyHost()
InboundRequestLoggingMiddleware

Pure-ASGI middleware that logs inbound HTTP requests.

Unlike BaseHTTPMiddleware, this passes the receive callable through to the inner application untouched, preserving request.is_disconnected() behaviour.

Wired automatically by AgentServerHost so that all protocol hosts (responses, invocations, etc.) get consistent inbound logging.
RequestIdMiddleware

Pure-ASGI middleware that sets x-request-id on every HTTP response.

The resolved request ID is stored in scope["state"] under <xref:REQUEST_ID_STATE_KEY> so that protocol-specific code (e.g. the Responses package) can read it for error body enrichment.

Unlike BaseHTTPMiddleware, this passes the receive callable through to the inner application untouched, preserving request.is_disconnected() behaviour.

Functions

build_server_version

Build a standard version segment for the x-platform-server header.

Format: {sdk_name}/{version} (python/{major}.{minor})

Protocol packages call this during host initialisation and pass the result to register_server_version.

build_server_version(sdk_name: str, version: str) -> str

Parameters

Name Description
sdk_name
Required
str

The SDK identifier (e.g., "azure-ai-agentserver-responses").
version
Required
str

The package version string (e.g., "1.0.0b1").

Returns

Type Description
str

A formatted version string.

Exceptions

Type Description
ValueError

If sdk_name or version is empty.

configure_observability

Default observability setup: console logging + tracing/OTel export.

Attaches a formatted StreamHandler to the root logger so that both SDK and user logging.info() calls are visible on the console. Then configures OpenTelemetry tracing and log export (Azure Monitor and/or OTLP) when a connection string or OTLP endpoint is available.

Pass this function (or a custom replacement) as the configure_observability parameter of AgentServerHost. Pass None to disable all SDK-managed observability setup.

configure_observability(*, connection_string: str | None = None, log_level: str | None = None) -> None

Keyword-Only Parameters

Name Description
connection_string
str or None

Application Insights connection string.

Default value: None
log_level
str or None

Log level name (e.g. "INFO", "DEBUG").

Default value: None

create_error_response

Build a JSONResponse with the standard error envelope.

create_error_response(code: str, message: str, *, status_code: int, error_type: str | None = None, details: list[dict[str, Any]] | None = None, headers: dict[str, str] | None = None) -> JSONResponse

Parameters

Name Description
code
Required
str

Machine-readable error code (e.g. "internal_error").
message
Required
str

Human-readable error message.

Keyword-Only Parameters

Name Description
status_code
int

HTTP status code for the response.
error_type
Optional[str]

Optional error type classification string. When provided, included as "type" in the error body.

Default value: None
details
Optional[list[dict[str, Any]]]

Child error objects, each with at least code and message keys.

Default value: None
headers
Optional[dict[str, str]]

Extra HTTP headers to include on the response.

Default value: None

Returns

Type Description
<xref:JSONResponse>

A ready-to-send JSON error response.

detach_context

Detach a context previously attached by set_current_span.

Best-effort no-op when token is None or when the token is no longer the current OpenTelemetry context.

detach_context(token: Any) -> None

Parameters

Name Description
token
Required
Any

The token returned by set_current_span.

end_span

End a span, optionally recording an error first.

No-op when span is None.

end_span(span: Any, exc: BaseException | None = None) -> None

Parameters

Name Description
span
Required
any

The OTel span to end, or None.
exc
BaseException or None

Optional exception to record before ending.

Default value: None

flush_spans

Flush all pending spans from the TracerProvider.

The BatchSpanProcessor buffers spans and exports them on a timer (default 5 seconds). In hosted sandbox environments the platform may suspend the process immediately after an HTTP response is sent, before the batch timer fires. Calling this function after ending the request span ensures that all child spans — including short-lived ones created by third-party tracers (e.g. LangChain/LangGraph per-node spans) — are exported before the sandbox is frozen.

No-op when the OTel SDK is not installed or the provider does not support force_flush.

flush_spans(timeout_millis: int = 5000) -> None

Parameters

Name Description
timeout_millis
int

Maximum time to wait for the flush, in milliseconds. Defaults to 5000 (5 seconds).

Default value: 5000

record_error

Record an exception and ERROR status on a span.

Sets error.type and otel.status.description per OTel semantic conventions.

record_error(span: Any, exc: BaseException) -> None

Parameters

Name Description
span
Required
any

The OTel span, or None.
exc
Required
BaseException

The exception to record.

set_current_span

Set a span as the current span in the OTel context.

This makes span the active parent for any child spans created by downstream code (e.g. framework handlers). Without this, spans created inside the handler would become siblings rather than children of span.

Returns a context token that must be passed to detach_context when the scope ends. No-op when span is None.

set_current_span(span: Any) -> Any

Parameters

Name Description
span
Required
Any

The OTel span to make current, or None.

Returns

Type Description
Any

A context token, or None.

trace_stream

Wrap a streaming body so the span covers the full transmission.

Yields chunks unchanged. Ends the span when the iterator is exhausted or raises an exception.

async trace_stream(iterator: AsyncIterable[str | bytes | memoryview], span: Any) -> AsyncIterator[str | bytes | memoryview]

Parameters

Name Description
iterator
Required
AsyncIterable[str or bytes or memoryview]

The async iterable to wrap.
span
Required
any

The OTel span to end on completion, or None.

Returns

Type Description
AsyncIterator[str,
bytes,
memoryview]

An async iterator yielding chunks unchanged.