AgentServerHost Class

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()

Constructor

AgentServerHost(*, applicationinsights_connection_string: str | None = None, graceful_shutdown_timeout: int | None = None, log_level: str | None = None, access_log: ~logging.Logger | None = <object object>, access_log_format: str | None = None, configure_observability: ~collections.abc.Callable[[...], None] | None = <function configure_observability>, routes: list[starlette.routing.Route] | None = None, **kwargs: ~typing.Any)

Parameters

Name Description
applicationinsights_connection_string
Required
Optional[str]

Application Insights connection string for exporting traces and logs to Azure Monitor. When None (default) the APPLICATIONINSIGHTS_CONNECTION_STRING env var is consulted. Tracing is automatically enabled when a connection string is available.
graceful_shutdown_timeout
Required
Optional[int]

Seconds to wait for in-flight requests to complete after receiving SIGTERM / shutdown signal. When None (default) the default is 30 seconds. Set to 0 to disable the drain period.
log_level
Required
Optional[str]

Library log level (e.g. "DEBUG", "INFO"). When None (default) the default is "INFO".
access_log
Required
Optional[Logger]

Logger for HTTP access logs, or None to disable. Defaults to the library logger (azure.ai.agentserver).
access_log_format
Required
Optional[str]

Hypercorn access-log format string. Supports %(h)s (remote addr), %(r)s (request line), %(s)s (status), %(b)s (body size), %(D)s (duration μs), %({header}i)s (request header), %({header}o)s (response header). Defaults to %(h)s "%(r)s" %(s)s %(b)s %(D)sμs.
configure_observability
Required
Optional[Callable[..., None]]

Callable that sets up console logging, tracing, and OTel export. Defaults to <xref:_tracing.configure_observability> which attaches a console handler to the root logger and configures Azure Monitor / OTLP export. Pass a custom callable to override the setup, or None to skip all SDK-managed observability configuration.

Keyword-Only Parameters

Name Description
applicationinsights_connection_string
Default value: None
graceful_shutdown_timeout
Default value: None
log_level
Default value: None
access_log
Default value: <object object at 0x000001B57D072330>
access_log_format
Default value: None
configure_observability
Default value: <function configure_observability at 0x000001B57F52B920>
routes
Default value: None

Methods

add_exception_handler
add_middleware
add_route
build_middleware_stack
host
mount
register_server_version

Register a version segment for the x-platform-server header.

Protocol packages (e.g. responses, invocations) call this in their __init__ to add their own portion. Handler developers can also call it to append a custom version string. Duplicates are ignored.

Use build_server_version to build a standard segment:


   from azure.ai.agentserver.core import build_server_version

   app.register_server_version(
       build_server_version("my-library", "2.0.0")
   )
request_span

Create a request-scoped span with this host's identity attributes.

Delegates to <xref:_tracing.request_span> with pre-populated agent identity from environment variables.
run

Start the server synchronously.
run_async

Start the server asynchronously (awaitable).
shutdown_handler

Register a function as the shutdown handler.
sse_keepalive_stream

Interleave SSE keep-alive comment frames into a streaming body.

Emits b": keep-alive\n\n" whenever the upstream iterator has not produced a chunk within interval seconds. This prevents proxies/load-balancers from closing idle connections.
url_path_for

add_exception_handler 

add_exception_handler(exc_class_or_status_code: int | type[Exception], handler: ExceptionHandler) -> None

Parameters

Name Description
exc_class_or_status_code
Required
handler
Required

add_middleware 

add_middleware(middleware_class: ~starlette.middleware._MiddlewareFactory[~P], *args: ~typing.~P, **kwargs: ~typing.~P) -> None

Parameters

Name Description
middleware_class
Required

add_route 

add_route(path: str, route: Callable[[Request], Awaitable[Response] | Response], methods: list[str] | None = None, name: str | None = None, include_in_schema: bool = True) -> None

Parameters

Name Description
path
Required
route
Required
methods
Default value: None
name
Default value: None
include_in_schema
Default value: True

build_middleware_stack 

build_middleware_stack() -> Callable[[MutableMapping[str, Any], Callable[[], Awaitable[MutableMapping[str, Any]]], Callable[[MutableMapping[str, Any]], Awaitable[None]]], Awaitable[None]]

host 

host(host: str, app: Callable[[MutableMapping[str, Any], Callable[[], Awaitable[MutableMapping[str, Any]]], Callable[[MutableMapping[str, Any]], Awaitable[None]]], Awaitable[None]], name: str | None = None) -> None

Parameters

Name Description
host
Required
app
Required
name
Default value: None

mount 

mount(path: str, app: Callable[[MutableMapping[str, Any], Callable[[], Awaitable[MutableMapping[str, Any]]], Callable[[MutableMapping[str, Any]], Awaitable[None]]], Awaitable[None]], name: str | None = None) -> None

Parameters

Name Description
path
Required
app
Required
name
Default value: None

register_server_version

Register a version segment for the x-platform-server header.

Protocol packages (e.g. responses, invocations) call this in their __init__ to add their own portion. Handler developers can also call it to append a custom version string. Duplicates are ignored.

Use build_server_version to build a standard segment:


   from azure.ai.agentserver.core import build_server_version

   app.register_server_version(
       build_server_version("my-library", "2.0.0")
   )

register_server_version(version_segment: str) -> None

Parameters

Name Description
version_segment
Required
str

The version string to register.

Exceptions

Type Description
ValueError

If version_segment is empty or whitespace-only.

request_span

Create a request-scoped span with this host's identity attributes.

Delegates to <xref:_tracing.request_span> with pre-populated agent identity from environment variables.

request_span(headers: Any, request_id: str, operation: str, *, operation_name: str | None = None, session_id: str = '', end_on_exit: bool = True) -> Any

Parameters

Name Description
headers
Required
any

HTTP request headers.
request_id
Required
str

The request/invocation ID.
operation
Required
str

Span operation (e.g. "invoke_agent").

Keyword-Only Parameters

Name Description
operation_name
str or None

Optional gen_ai.operation.name value.

Default value: None
session_id
str

Session ID.
end_on_exit
bool

Whether to end the span when the context exits.

Default value: True

Returns

Type Description
any

Context manager yielding the OTel span.

run

Start the server synchronously.

run(host: str = '0.0.0.0', port: int | None = None) -> None

Parameters

Name Description
host
str

Network interface to bind. Defaults to "0.0.0.0".

Default value: 0.0.0.0
port
Optional[int]

Port to bind. Defaults to PORT env var or 8088.

Default value: None

run_async

Start the server asynchronously (awaitable).

async run_async(host: str = '0.0.0.0', port: int | None = None) -> None

Parameters

Name Description
host
str

Network interface to bind. Defaults to "0.0.0.0".

Default value: 0.0.0.0
port
Optional[int]

Port to bind. Defaults to PORT env var or 8088.

Default value: None

shutdown_handler

Register a function as the shutdown handler.

shutdown_handler(fn: Callable[[], Awaitable[None]]) -> Callable[[], Awaitable[None]]

Parameters

Name Description
fn
Required
Callable[[], Awaitable[None]]

Async function called during graceful shutdown.

Returns

Type Description
Callable[[], Awaitable[None]]

The original function (unmodified).

sse_keepalive_stream

Interleave SSE keep-alive comment frames into a streaming body.

Emits b": keep-alive\n\n" whenever the upstream iterator has not produced a chunk within interval seconds. This prevents proxies/load-balancers from closing idle connections.

static sse_keepalive_stream(iterator: AsyncIterable[str | bytes | memoryview], interval: int) -> AsyncIterator[str | bytes | memoryview]

Parameters

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

The async iterable to wrap.
interval
Required
int

Seconds between keep-alive frames. Must be > 0.

Returns

Type Description
AsyncIterator[str,
bytes,
memoryview]

An async iterator with interleaved keep-alive frames.

url_path_for 

url_path_for(name: str, /, **path_params: Any) -> URLPath

Positional-Only Parameters

Name Description
name
Required

Attributes

routes