InvocationAgentServerHost Class

Invocation protocol host for Azure AI Hosted Agents.

A AgentServerHost subclass that adds the invocation protocol endpoints. Use the decorator methods to wire handler functions to the endpoints.

For multi-protocol agents, compose via cooperative inheritance:


   class MyHost(InvocationAgentServerHost, ResponsesAgentServerHost):
       pass

Usage:


   from azure.ai.agentserver.invocations import InvocationAgentServerHost

   app = InvocationAgentServerHost()

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

   app.run()

Constructor

InvocationAgentServerHost(*, openapi_spec: dict[str, Any] | None = None, **kwargs: Any)

Parameters

Name Description
openapi_spec
Required
Optional[dict[str, Any]]

Optional OpenAPI spec dict. When provided, the spec is served at GET /invocations/docs/openapi.json.

Keyword-Only Parameters

Name Description
openapi_spec
Default value: None

Methods

add_exception_handler
add_middleware
add_route
build_middleware_stack
cancel_invocation_handler

Register a function as the cancel-invocation handler.
get_invocation_handler

Register a function as the get-invocation handler.
get_openapi_spec

Return the stored OpenAPI spec, or None.
host
invoke_handler

Register a function as the invoke handler.

Usage:


   @invocations.invoke_handler
   async def handle(request: Request) -> Response:
       ...
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]]

cancel_invocation_handler

Register a function as the cancel-invocation handler.

cancel_invocation_handler(fn: Callable[[Request], Awaitable[Response]]) -> Callable[[Request], Awaitable[Response]]

Parameters

Name Description
fn
Required
Callable[[<xref:Request>], Awaitable[<xref:Response>]]

Async function accepting a Starlette Request and returning a Response.

Returns

Type Description
Callable[[<xref:Request>], Awaitable[<xref:Response>]]

The original function (unmodified).

Exceptions

Type Description
TypeError

If fn is not an async function.

get_invocation_handler

Register a function as the get-invocation handler.

get_invocation_handler(fn: Callable[[Request], Awaitable[Response]]) -> Callable[[Request], Awaitable[Response]]

Parameters

Name Description
fn
Required
Callable[[<xref:Request>], Awaitable[<xref:Response>]]

Async function accepting a Starlette Request and returning a Response.

Returns

Type Description
Callable[[<xref:Request>], Awaitable[<xref:Response>]]

The original function (unmodified).

Exceptions

Type Description
TypeError

If fn is not an async function.

get_openapi_spec

Return the stored OpenAPI spec, or None.

get_openapi_spec() -> dict[str, Any] | 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

invoke_handler

Register a function as the invoke handler.

Usage:


   @invocations.invoke_handler
   async def handle(request: Request) -> Response:
       ...

invoke_handler(fn: Callable[[Request], Awaitable[Response]]) -> Callable[[Request], Awaitable[Response]]

Parameters

Name Description
fn
Required
Callable[[<xref:Request>], Awaitable[<xref:Response>]]

Async function accepting a Starlette Request and returning a Response.

Returns

Type Description
Callable[[<xref:Request>], Awaitable[<xref:Response>]]

The original function (unmodified).

Exceptions

Type Description
TypeError

If fn is not an async function.

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.

async 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