Azure · ZhaoChaoqun · Jul 3, 2026 · Jul 6, 2026
@@ -1,5 +1,15 @@
 # Release History
 
+## 1.0.0b7 (Unreleased)
+
+### Features Added
+
+- AsyncAPI docs endpoints — `InvocationAgentServerHost` now accepts optional
+  `asyncapi_spec_json` (dict) and/or `asyncapi_spec_yaml` (raw YAML string)
+  constructor args, served at `GET /invocations/docs/asyncapi.json` and
+  `GET /invocations/docs/asyncapi.yaml` respectively. Either representation
+  returns `404` if not registered. See README for details.
+
 ## 1.0.0b6 (2026-06-28)
 
 ### Features Added

@@ -2,7 +2,7 @@
 
 The `azure-ai-agentserver-invocations` package provides the invocation protocol endpoints for Azure AI Hosted Agent containers. It plugs into the [`azure-ai-agentserver-core`](https://pypi.org/project/azure-ai-agentserver-core/) host framework and supports two transports on the same host:
 
-- **HTTP** (`invocations` protocol) — `POST /invocations`, `GET /invocations/{id}`, `POST /invocations/{id}/cancel`, `GET /invocations/docs/openapi.json`.
+- **HTTP** (`invocations` protocol) — `POST /invocations`, `GET /invocations/{id}`, `POST /invocations/{id}/cancel`, `GET /invocations/docs/openapi.json`, `GET /invocations/docs/asyncapi.{json,yaml}`.
 - **WebSocket** (`invocations_ws` protocol) — full-duplex streaming at `/invocations_ws`, registered with `@app.ws_handler`.
 
 ## Getting started
@@ -38,6 +38,8 @@ This automatically installs `azure-ai-agentserver-core` as a dependency.
 | `GET` | `/invocations/{invocation_id}` | No | Retrieve invocation status or result |
 | `POST` | `/invocations/{invocation_id}/cancel` | No | Cancel a running invocation |
 | `GET` | `/invocations/docs/openapi.json` | No | Serve the agent's OpenAPI 3.x spec |
+| `GET` | `/invocations/docs/asyncapi.json` | No | Serve the agent's AsyncAPI 3.x spec (JSON) |
+| `GET` | `/invocations/docs/asyncapi.yaml` | No | Serve the agent's AsyncAPI 3.x spec (YAML) |
 | `WS`   | `/invocations_ws` | No | Full-duplex WebSocket transport (`invocations_ws` protocol) |
 
 ### Request and response headers
@@ -225,6 +227,39 @@ app = InvocationAgentServerHost(openapi_spec={
 })
 ```
 
+### Serving an AsyncAPI spec
+
+AsyncAPI is the companion to OpenAPI for streaming/bidirectional surfaces (e.g. the
+`invocations_ws` WebSocket protocol) that OpenAPI cannot express. Pass either or both
+representations to enable the discovery endpoints:
+
+```python
+app = InvocationAgentServerHost(
+    asyncapi_spec_json={
+        "asyncapi": "3.0.0",
+        "info": {"title": "My Agent", "version": "1.0.0"},
+        "channels": { ... },
+        "operations": { ... },
+    },
+    asyncapi_spec_yaml="""asyncapi: 3.0.0
+info:
+  title: My Agent
+  version: 1.0.0
+channels:
+  ...
+""",
+)
+```
+
+Each representation is served at its own path:
+
+- `GET /invocations/docs/asyncapi.json` — `application/json`
+- `GET /invocations/docs/asyncapi.yaml` — `application/yaml`
+
+The path extension is authoritative for the returned content type (no `Accept`
+negotiation, no format conversion). If you only pass one, the other returns `404`.
+Serving both is recommended for tooling compatibility.
+
 ## WebSocket protocol (`invocations_ws`)
 
 The same `InvocationAgentServerHost` object also exposes a WebSocket transport at `/invocations_ws`. Container authors do not install or import a second package — registering an `@app.ws_handler` is the only step. A multi-protocol agent shares one host, one session, and one process.

@@ -16,6 +16,8 @@ namespace azure.ai.agentserver.invocations
                 self, 
                 *, 
                 openapi_spec: Optional[dict[str, Any]] = ..., 
+                asyncapi_spec_json: Optional[dict[str, Any]] = ..., 
+                asyncapi_spec_yaml: Optional[str] = ..., 
                 **kwargs: Any
             ) -> None: ...
 
@@ -45,6 +47,10 @@ namespace azure.ai.agentserver.invocations
 
         def cancel_invocation_handler(self, fn: Callable[[Request], Awaitable[Response]]) -> Callable[[Request], Awaitable[Response]]: ...
 
+        def get_asyncapi_spec_json(self) -> Optional[dict[str, Any]]: ...
+
+        def get_asyncapi_spec_yaml(self) -> Optional[str]: ...
+
         def get_invocation_handler(self, fn: Callable[[Request], Awaitable[Response]]) -> Callable[[Request], Awaitable[Response]]: ...
 
         def get_openapi_spec(self) -> Optional[dict[str, Any]]: ...

@@ -184,6 +184,15 @@ async def ws(websocket: WebSocket) -> None:
     :param openapi_spec: Optional OpenAPI spec dict.  When provided, the spec
         is served at ``GET /invocations/docs/openapi.json``.
     :type openapi_spec: Optional[dict[str, Any]]
+    :param asyncapi_spec_json: Optional AsyncAPI spec dict.  When provided, the spec
+        is served at ``GET /invocations/docs/asyncapi.json``.  AsyncAPI is the companion
+        to OpenAPI for streaming/bidirectional surfaces (e.g. ``invocations_ws``).
+    :type asyncapi_spec_json: Optional[dict[str, Any]]
+    :param asyncapi_spec_yaml: Optional AsyncAPI spec as raw YAML text.  When provided,
+        the spec is served verbatim at ``GET /invocations/docs/asyncapi.yaml``.  The
+        path extension is authoritative for the returned content type (no ``Accept``
+        negotiation); the platform never converts between JSON and YAML.
+    :type asyncapi_spec_yaml: Optional[str]
     """
 
     _INSTRUMENTATION_SCOPE = "Azure.AI.AgentServer.Invocations"
@@ -192,12 +201,29 @@ def __init__(
         self,
         *,
         openapi_spec: Optional[dict[str, Any]] = None,
+        asyncapi_spec_json: Optional[dict[str, Any]] = None,
+        asyncapi_spec_yaml: Optional[str] = None,
         **kwargs: Any,
     ) -> None:
         self._invoke_fn: Optional[Callable] = None
         self._get_invocation_fn: Optional[Callable] = None
         self._cancel_invocation_fn: Optional[Callable] = None
+        # asyncapi_spec_* are validated eagerly: a stringified dict passed as
+        # asyncapi_spec_yaml is technically valid YAML and would silently reach
+        # clients as a single-line JSON blob. openapi_spec is left unchecked to
+        # preserve backward-compat with existing callers passing dict-like
+        # Mapping subclasses; consider unifying in a future release.
+        if asyncapi_spec_json is not None and not isinstance(asyncapi_spec_json, dict):
+            raise TypeError(
+                f"asyncapi_spec_json must be dict, got {type(asyncapi_spec_json).__name__}"
+            )
+        if asyncapi_spec_yaml is not None and not isinstance(asyncapi_spec_yaml, str):
+            raise TypeError(
+                f"asyncapi_spec_yaml must be str, got {type(asyncapi_spec_yaml).__name__}"
+            )
         self._openapi_spec = openapi_spec
+        self._asyncapi_spec_json = asyncapi_spec_json
+        self._asyncapi_spec_yaml = asyncapi_spec_yaml
 
         # Initialise WS handler slots (no parameters — the keep-alive
         # interval lives on ``AgentConfig`` and is wired into Hypercorn
@@ -212,6 +238,18 @@ def __init__(
                 methods=["GET"],
                 name="get_openapi_spec",
             ),
+            Route(
+                "/invocations/docs/asyncapi.json",
+                self._get_asyncapi_spec_json_endpoint,
+                methods=["GET"],
+                name="get_asyncapi_spec_json",
+            ),
+            Route(
+                "/invocations/docs/asyncapi.yaml",
+                self._get_asyncapi_spec_yaml_endpoint,
+                methods=["GET"],
+                name="get_asyncapi_spec_yaml",
+            ),
             Route(
                 "/invocations",
                 self._create_invocation_endpoint,
@@ -238,8 +276,11 @@ def __init__(
 
         # --- Invocations startup configuration logging ---
         logger.info(
-            "Invocations protocol: openapi_spec_configured=%s",
+            "Invocations protocol: openapi_spec_configured=%s, "
+            "asyncapi_spec_json_configured=%s, asyncapi_spec_yaml_configured=%s",
             self._openapi_spec is not None,
+            self._asyncapi_spec_json is not None,
+            self._asyncapi_spec_yaml is not None,
         )
 
     # ------------------------------------------------------------------
@@ -342,6 +383,14 @@ def get_openapi_spec(self) -> Optional[dict[str, Any]]:
         """Return the stored OpenAPI spec, or None."""
         return self._openapi_spec
 
+    def get_asyncapi_spec_json(self) -> Optional[dict[str, Any]]:
+        """Return the stored AsyncAPI spec (JSON representation), or None."""
+        return self._asyncapi_spec_json
+
+    def get_asyncapi_spec_yaml(self) -> Optional[str]:
+        """Return the stored AsyncAPI spec as raw YAML text, or None."""
+        return self._asyncapi_spec_yaml
+
     # ------------------------------------------------------------------
     # Span attribute helper
     # ------------------------------------------------------------------
@@ -360,6 +409,26 @@ async def _get_openapi_spec_endpoint(self, request: Request) -> Response:  # pyl
             )
         return JSONResponse(spec)
 
+    async def _get_asyncapi_spec_json_endpoint(self, request: Request) -> Response:  # pylint: disable=unused-argument
+        spec = self.get_asyncapi_spec_json()
+        if spec is None:
+            return create_error_response(
+                "not_found", "No AsyncAPI (JSON) spec registered",
+                status_code=404,
+                headers=_apply_error_source_headers({}, _ERROR_SOURCE_UPSTREAM),
+            )
+        return JSONResponse(spec)
+
+    async def _get_asyncapi_spec_yaml_endpoint(self, request: Request) -> Response:  # pylint: disable=unused-argument
+        spec = self.get_asyncapi_spec_yaml()
+        if spec is None:
+            return create_error_response(
+                "not_found", "No AsyncAPI (YAML) spec registered",
+                status_code=404,
+                headers=_apply_error_source_headers({}, _ERROR_SOURCE_UPSTREAM),
+            )
+        return Response(spec, media_type="application/yaml; charset=utf-8")
+
     def _wrap_streaming_response(
         self,
         response: StreamingResponse,

@@ -2,4 +2,4 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # ---------------------------------------------------------
 
-VERSION = "1.0.0b6"
+VERSION = "1.0.0b7"
@@ -69,6 +69,116 @@ async def handle(request: Request) -> Response:
     assert resp.json() == SAMPLE_OPENAPI_SPEC
 
 
+# ---------------------------------------------------------------------------
+# GET asyncapi specs return 404 when not set
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_get_asyncapi_json_returns_404_when_not_set(no_spec_client):
+    """GET /invocations/docs/asyncapi.json returns 404 when no spec registered."""
+    resp = await no_spec_client.get("/invocations/docs/asyncapi.json")
+    assert resp.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_get_asyncapi_yaml_returns_404_when_not_set(no_spec_client):
+    """GET /invocations/docs/asyncapi.yaml returns 404 when no spec registered."""
+    resp = await no_spec_client.get("/invocations/docs/asyncapi.yaml")
+    assert resp.status_code == 404
+
+
+# ---------------------------------------------------------------------------
+# GET asyncapi specs return spec when registered
+# ---------------------------------------------------------------------------
+
+SAMPLE_ASYNCAPI_SPEC_JSON = {
+    "asyncapi": "3.0.0",
+    "info": {"title": "Test Agent", "version": "1.0.0"},
+    "channels": {},
+    "operations": {},
+}
+
+SAMPLE_ASYNCAPI_SPEC_YAML = (
+    "asyncapi: 3.0.0\n"
+    "info:\n"
+    "  title: Test Agent\n"
+    "  version: 1.0.0\n"
+    "channels: {}\n"
+    "operations: {}\n"
+)
+
+
+@pytest.mark.asyncio
+async def test_get_asyncapi_json_returns_spec_when_registered():
+    """GET /invocations/docs/asyncapi.json returns the spec when registered."""
+    app = InvocationAgentServerHost(asyncapi_spec_json=SAMPLE_ASYNCAPI_SPEC_JSON)
+
+    @app.invoke_handler
+    async def handle(request: Request) -> Response:
+        return Response(content=b"ok")
+
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://testserver") as client:
+        resp = await client.get("/invocations/docs/asyncapi.json")
+    assert resp.status_code == 200
+    assert resp.json() == SAMPLE_ASYNCAPI_SPEC_JSON
+
+
+@pytest.mark.asyncio
+async def test_get_asyncapi_yaml_returns_spec_when_registered():
+    """GET /invocations/docs/asyncapi.yaml returns the spec verbatim with YAML media type."""
+    app = InvocationAgentServerHost(asyncapi_spec_yaml=SAMPLE_ASYNCAPI_SPEC_YAML)
+
+    @app.invoke_handler
+    async def handle(request: Request) -> Response:
+        return Response(content=b"ok")
+
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://testserver") as client:
+        resp = await client.get("/invocations/docs/asyncapi.yaml")
+    assert resp.status_code == 200
+    assert resp.text == SAMPLE_ASYNCAPI_SPEC_YAML
+    assert resp.headers["content-type"].startswith("application/yaml")
+
+
+@pytest.mark.asyncio
+async def test_asyncapi_representations_are_independent():
+    """Registering only JSON leaves YAML as 404 and vice versa (no format conversion)."""
+    json_only = InvocationAgentServerHost(asyncapi_spec_json=SAMPLE_ASYNCAPI_SPEC_JSON)
+
+    @json_only.invoke_handler
+    async def handle_json_only(request: Request) -> Response:
+        return Response(content=b"ok")
+
+    transport = ASGITransport(app=json_only)
+    async with AsyncClient(transport=transport, base_url="http://testserver") as client:
+        assert (await client.get("/invocations/docs/asyncapi.json")).status_code == 200
+        assert (await client.get("/invocations/docs/asyncapi.yaml")).status_code == 404
+
+    yaml_only = InvocationAgentServerHost(asyncapi_spec_yaml=SAMPLE_ASYNCAPI_SPEC_YAML)
+
+    @yaml_only.invoke_handler
+    async def handle_yaml_only(request: Request) -> Response:
+        return Response(content=b"ok")
+
+    transport = ASGITransport(app=yaml_only)
+    async with AsyncClient(transport=transport, base_url="http://testserver") as client:
+        assert (await client.get("/invocations/docs/asyncapi.yaml")).status_code == 200
+        assert (await client.get("/invocations/docs/asyncapi.json")).status_code == 404
+
+
+def test_asyncapi_json_rejects_non_dict():
+    """Constructor rejects non-dict asyncapi_spec_json to avoid silent misuse."""
+    with pytest.raises(TypeError, match="asyncapi_spec_json must be dict"):
+        InvocationAgentServerHost(asyncapi_spec_json="not a dict")  # type: ignore[arg-type]
+
+
+def test_asyncapi_yaml_rejects_non_str():
+    """Constructor rejects non-str asyncapi_spec_yaml to avoid stringified dict leaking to client."""
+    with pytest.raises(TypeError, match="asyncapi_spec_yaml must be str"):
+        InvocationAgentServerHost(asyncapi_spec_yaml={"not": "a string"})  # type: ignore[arg-type]
+
+
 # ---------------------------------------------------------------------------
 # GET /invocations/{id} returns 404 default
 # ---------------------------------------------------------------------------