[agentserver-invocations] Serve AsyncAPI docs at /invocations/docs/asyncapi.{json,yaml} (#47829)

* [agentserver-invocations] Serve AsyncAPI docs at /invocations/docs/asyncapi.{json,yaml}

InvocationAgentServerHost now accepts optional `asyncapi_spec_json` (dict) and/or
`asyncapi_spec_yaml` (raw YAML str) constructor args. When set, they are served
at GET /invocations/docs/asyncapi.json and GET /invocations/docs/asyncapi.yaml
respectively — the AsyncAPI companion to the existing openapi.json endpoint for
streaming/bidirectional surfaces (e.g. the invocations_ws WebSocket protocol)
that OpenAPI cannot express.

Path extension is authoritative for the returned content type — no Accept
negotiation, no format conversion between the two representations. Either
returns 404 if not registered; users may register only one, only the other,
or both.

Constructor rejects wrong-typed args (dict/str) with TypeError to prevent
misuse from silently leaking a stringified dict to clients as YAML.

Startup log extended to record which specs are configured.

Bumps version to 1.0.0b7 and syncs api.md with the new public surface.

Companion to TypeSpec PR Azure/azure-rest-api-specs#44416.

Fixes: AB#5407709

* Address review: yaml charset, drop redundant asyncio marks, clarify validation

- _invocation.py: set media_type to "application/yaml; charset=utf-8".
  RFC 9512 mandates UTF-8 for application/yaml, but clients don't
  universally assume it; declaring the charset avoids downstream
  mis-decoding of non-ASCII content in the raw YAML string.
- _invocation.py: add a comment on the asyncapi_spec_* isinstance
  guards explaining why openapi_spec is intentionally left unchecked
  (backward-compat with dict-like Mapping subclasses), so future
  readers don't file the inconsistency as a bug.
- test_server_routes.py: drop @pytest.mark.asyncio + async on the two
  TypeError-raising tests. Their bodies are purely synchronous
  (constructor call inside pytest.raises); the async wrapping was
  spurious.

Author: ZhaoChaoqun

sdk/agentserver/azure-ai-agentserver-invocations/CHANGELOG.md

@@ -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

sdk/agentserver/azure-ai-agentserver-invocations/README.md

@@ -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.

sdk/agentserver/azure-ai-agentserver-invocations/api.md

@@ -15,6 +15,8 @@ namespace azure.ai.agentserver.invocations
         def __init__(
                 self, 
                 *, 
+                asyncapi_spec_json: Optional[dict[str, Any]] = ..., 
+                asyncapi_spec_yaml: Optional[str] = ..., 
                 openapi_spec: Optional[dict[str, Any]] = ..., 
                 **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]]: ...

sdk/agentserver/azure-ai-agentserver-invocations/api.metadata.yml

@@ -1,3 +1,3 @@
-apiMdSha256: 20e0c715368a99cb4302338429be892247a5f77a7500a96cf5164fbce402050d
+apiMdSha256: dbfee3ed6344b859cdcfa55da4772aa2cdb8e4316bbc99f9bc8c7c2abf43e00e
 parserVersion: 0.3.28
-pythonVersion: 3.11.9
+pythonVersion: 3.14.2

sdk/agentserver/azure-ai-agentserver-invocations/azure/ai/agentserver/invocations/_invocation.py

@@ -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,

sdk/agentserver/azure-ai-agentserver-invocations/azure/ai/agentserver/invocations/_version.py

@@ -2,4 +2,4 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # ---------------------------------------------------------
 
-VERSION = "1.0.0b6"
+VERSION = "1.0.0b7"

sdk/agentserver/azure-ai-agentserver-invocations/tests/test_server_routes.py

@@ -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
 # ---------------------------------------------------------------------------