azure-ai-agentserver-core 2.0.0b8

Foundation utilities and host framework for Azure AI Hosted Agents

Azure AI Agent Server Core client library for Python

The azure-ai-agentserver-core package provides the foundation host framework for building Azure AI Hosted Agent containers. It handles the protocol-agnostic infrastructure — health probes, graceful shutdown, OpenTelemetry tracing, and ASGI serving — so that protocol packages can focus on their endpoint logic.

Getting started

Install the package

pip install azure-ai-agentserver-core

OpenTelemetry tracing with Azure Monitor and OTLP exporters is included by default.

Prerequisites

• Python 3.10 or later

Key concepts

AgentServerHost

AgentServerHost is the host process for Azure AI Hosted Agent containers. It provides:

• Health probe — GET /readiness returns 200 OK when the server is ready.
• Graceful shutdown — On SIGTERM the server drains in-flight requests (default 30 s timeout) before exiting.
• OpenTelemetry tracing — Automatic span creation with Azure Monitor and OTLP export when configured.
• Hypercorn ASGI server — Serves on 0.0.0.0:${PORT:-8088} with HTTP/1.1.

Protocol packages (e.g. azure-ai-agentserver-invocations) subclass AgentServerHost and add their endpoints in __init__.

Environment variables

Variable	Description	Default
PORT	Listen port	8088
FOUNDRY_AGENT_NAME	Agent name (used in tracing)	""
FOUNDRY_AGENT_VERSION	Agent version (used in tracing)	""
FOUNDRY_PROJECT_ENDPOINT	Azure AI Foundry project endpoint	""
FOUNDRY_PROJECT_ARM_ID	Foundry project ARM resource ID (used in tracing)	""
FOUNDRY_AGENT_SESSION_ID	Default session ID when not provided per-request	""
APPLICATIONINSIGHTS_CONNECTION_STRING	Azure Monitor connection string	—
OTEL_EXPORTER_OTLP_ENDPOINT	OTLP collector endpoint	—

Examples

AgentServerHost is typically used via a protocol package. The simplest setup with the invocations protocol:

from azure.ai.agentserver.invocations import InvocationAgentServerHost
from starlette.responses import JSONResponse

app = InvocationAgentServerHost()

@app.invoke_handler
async def handle(request):
    body = await request.json()
    return JSONResponse({"greeting": f"Hello, {body['name']}!"})

app.run()

Per-request identity (multi-user sessions)

On container protocol 2.0.0 a single agent session can serve multiple users. Each request carries x-agent-user-id (the user — partition state by it) and an opaque x-agent-foundry-call-id (the per-request caller identity). Read both via get_request_context(); the SDK forwards only the call ID on outbound Foundry calls — x-agent-user-id is never echoed. Forwarding the call ID lets a tool server resolve which user made the request and act on their behalf.

import os

import httpx
from azure.ai.agentserver.core import get_request_context


def foundry_headers() -> dict[str, str]:
    # Echoes x-agent-foundry-call-id only; x-agent-user-id is never forwarded.
    return dict(get_request_context().platform_headers())


async def call_toolbox(query: str) -> str:
    user_id = get_request_context().user_id  # for the container's OWN per-user state
    # Attach the call ID PER CALL — a toolbox MCP session is long-lived and serves many
    # users/turns, so never bake one call's ID into the client's static 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()}", **foundry_headers()},
            json={"jsonrpc": "2.0", "method": "tools/call",
                  "params": {"name": "list_my_assigned_issues", "arguments": {}}},
        )
    return resp.text  # the toolbox resolved the caller from the call ID and acted as that user

Subclassing AgentServerHost

For custom protocol implementations, subclass AgentServerHost and add routes:

from azure.ai.agentserver.core import AgentServerHost
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Route

class MyAgentHost(AgentServerHost):
    def __init__(self, **kwargs):
        my_routes = [Route("/my-endpoint", self._handle, methods=["POST"])]
        existing = list(kwargs.pop("routes", None) or [])
        super().__init__(routes=existing + my_routes, **kwargs)

    async def _handle(self, request: Request):
        return JSONResponse({"status": "ok"})

app = MyAgentHost()
app.run()

Shutdown handler

Register a cleanup function that runs during graceful shutdown:

app = AgentServerHost()

@app.shutdown_handler
async def on_shutdown():
    # Close database connections, flush buffers, etc.
    pass

Durable state storage

FoundryStateStore is a durable, server-backed key-value store for agent state — session memory, per-user preferences, counters, and checkpoints — bound to one explicit, caller-named store, with single-item optimistic concurrency, tag-filtered key listing, and store-level TTL.

from azure.ai.agentserver.core.storage import FoundryStateStore

# Endpoint and credential resolve from FOUNDRY_PROJECT_ENDPOINT + DefaultAzureCredential.
# get_or_create() resolves (or creates, on first use) the store in one call.
store = await FoundryStateStore.get_or_create("checkpoints/thread-abc", user_isolation=True)
async with store:
    await store.set_item("step-1", {"done": False})
    item = await store.get_item("step-1")
    print(item.value)  # {"done": False}

The default credential path uses DefaultAzureCredential, which requires the optional azure-identity package (pip install azure-identity). Alternatively, pass any azure.core.credentials_async.AsyncTokenCredential explicitly to get_or_create() and azure-identity is not needed.

Reads return typed StateStoreItem values; writes return typed item metadata and use single-item If-Match concurrency. Session/conversation scoping is expressed in the store name itself, and item expiry is controlled by the store's item_ttl_seconds setting. See the Durable State Store Guide for the full API, the store lifecycle, and common gotchas, and state_store_sample.py for a runnable end-to-end example.

Configuring tracing

Tracing is enabled automatically when an Application Insights connection string is available:

app = AgentServerHost(
    applicationinsights_connection_string="InstrumentationKey=...",
)

Or via environment variable:

export APPLICATIONINSIGHTS_CONNECTION_STRING="InstrumentationKey=..."
python my_agent.py

Troubleshooting

Logging

Set the log level to DEBUG for detailed diagnostics:

app = AgentServerHost(log_level="DEBUG")

Reporting issues

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

Next steps

• Install azure-ai-agentserver-invocations to add the invocation protocol endpoints.
• See the container image spec for the full hosted agent contract.

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 [e-mail] with any additional questions or comments.