feat: Add Try It Out feature for AsyncAPI documentation

docs/docs/en/getting-started/asgi.md

@@ -91,7 +91,7 @@ If you want to write your own simple **HTTP**-endpoint, you can use our `#!pytho
 
 You can also use helper functions to access query parameters and headers:
 
-```python linenums="1" hl_lines="1 8-9 19"
+```python linenums="1" hl_lines="1 8-9 18"
 {! docs_src/getting_started/asgi/auth_app.py !}
 ```
 
@@ -106,18 +106,19 @@ Dependency Injection works with [**FastDepends**](https://lancetnik.github.io/Fa
 
 By default, any ASGI routes will be added to your AsyncAPI documentation. If you wish to exclude these routes, just do the following:
 
-```python linenums="1"
+```python linenums="1" hl_lines="5"
 app = AsgiFastStream(
     broker,
-    asgi_routes=[
-        ("/health", make_ping_asgi(broker, timeout=5.0, include_in_schema=False)),
-    ]
+    asgi_routes=[(
+        "/health",
+        make_ping_asgi(broker, timeout=5.0, include_in_schema=False)
+    )]
 )
 ```
 
 Or, for custom ASGI routes:
 
-```python linenums="1"
+```python linenums="1" hl_lines="1"
 @get(include_in_schema=False)
 async def liveness_ping(scope):
     return AsgiResponse(b"", status_code=200)
@@ -148,17 +149,20 @@ app = AsgiFastStream(
 )
 ```
 
-Now, your **AsyncAPI HTML** representation can be found by the `/docs` url.
+Now, your **AsyncAPI HTML** representation can be found by the `/docs/asyncapi` url.
+
+!!! note
+    For extended examples on the **AsyncAPI** feature, see [Serving the AsyncAPI Documentation](./asyncapi/hosting.md){.internal-link} page.
 
 ### FastStream Object Reuse
 
-You may also use regular `FastStream` application object for similar result.
+You may also use regular `FastStream.as_asgi()` method for similar result.
 
-```python linenums="1" hl_lines="2 12"
+```python linenums="1" hl_lines="1 12"
 from faststream import FastStream
 from faststream.nats import NatsBroker
 from faststream.specification import AsyncAPI
-from faststream.asgi import make_ping_asgi, AsgiResponse
+from faststream.asgi import make_ping_asgi, AsgiResponse, get
 
 broker = NatsBroker()
 
@@ -211,5 +215,5 @@ async def start_broker(app):
 app = FastAPI(lifespan=start_broker)
 
 app.mount("/health", make_ping_asgi(broker, timeout=5.0))
-app.mount("/asyncapi", make_asyncapi_asgi(asyncapi))
+app.mount("/asyncapi", make_asyncapi_asgi(asyncapi, try_it_out=False))
 ```

docs/docs/en/getting-started/asyncapi/hosting.md

@@ -95,6 +95,59 @@ app = AsgiFastStream(
 
 After running the script, the **AsyncAPI** docs will be available at: <http://localhost:8000/docs/asyncapi>
 
+## Try It Out
+
+The AsyncAPI documentation page includes a built-in **Try It Out** feature that lets you publish test messages directly from the browser UI, without leaving the docs page.
+
+!!! note
+    The Try It Out UI is powered by the [asyncapi-try-it-plugin](https://github.com/Shepard2154/asyncapi-try-it-plugin){.external-link target="_blank"} — a plugin for AsyncAPI that adds a Swagger-like interface to send test messages to message brokers. Thanks to the maintainers for this functionality.
+
+By default, when you set `asyncapi_path`, a companion `POST` endpoint is automatically registered at `{asyncapi_path}/try`. The UI sends the message payload to this endpoint, which publishes it to your broker in test mode (without requiring a real broker connection).
+
+```python linenums="1" hl_lines="7"
+from faststream.nats import NatsBroker
+from faststream.asgi import AsgiFastStream
+
+broker = NatsBroker()
+
+# POST /docs/asyncapi/try is registered automatically
+app = AsgiFastStream(broker, asyncapi_path="/docs/asyncapi")
+```
+
+To disable the feature, use `AsyncAPIRoute` with `try_it_out=False`:
+
+```python linenums="1" hl_lines="2 8"
+from faststream.nats import NatsBroker
+from faststream.asgi import AsgiFastStream, AsyncAPIRoute
+
+broker = NatsBroker()
+
+app = AsgiFastStream(
+    broker,
+    asyncapi_path=AsyncAPIRoute("/docs/asyncapi", try_it_out=False),
+)
+```
+
+If you want to point the Try It Out UI to an **external backend** (e.g. a separate service or a production broker URL), pass a custom `try_it_out_url` via `AsyncAPIRoute`:
+
+```python linenums="1" hl_lines="2 10"
+from faststream.nats import NatsBroker
+from faststream.asgi import AsgiFastStream, AsyncAPIRoute
+
+broker = NatsBroker()
+
+app = AsgiFastStream(
+    broker,
+    asyncapi_path=AsyncAPIRoute(
+        "/docs/asyncapi",
+        try_it_out_url="https://api.example.com/asyncapi/try",
+    ),
+)
+```
+
+!!! note
+    When `try_it_out_url` is set on `AsyncAPIRoute`, it overrides the URL the browser sends requests to. The local `POST {asyncapi_path}/try` endpoint is still registered and reachable regardless of `try_it_out_url`, unless you also pass `try_it_out=False`.
+
 ## Integration with Different HTTP Frameworks (**FastAPI** Example)
 
 **FastStream** provides two robust approaches to combine your message broker documentation with any **ASGI** web frameworks.

faststream/_internal/testing/broker.py

@@ -94,12 +94,15 @@ async def _create_ctx(self) -> AsyncGenerator[Broker, None]:
 
         with context:
             async with self.broker:
+                saved_running = {sub: sub.running for sub in self.broker.subscribers}
                 try:
                     if not self.connect_only:
                         await self.broker.start()
                     yield self.broker
                 finally:
                     self._fake_close(self.broker)
+                    for sub, was_running in saved_running.items():
+                        sub.running = was_running
 
     @contextmanager
     def _patch_producer(self, broker: Broker) -> Iterator[None]:

faststream/asgi/app.py

@@ -17,7 +17,7 @@
 from faststream._internal.logger import logger
 from faststream.exceptions import INSTALL_UVICORN, StartupValidationError
 
-from .factories import AsyncAPIRoute
+from .factories import AsyncAPIRoute, make_try_it_out_handler
 from .handlers import HttpHandler
 from .response import AsgiResponse
 from .websocket import WebSocketClose
@@ -126,9 +126,23 @@ def __init__(
         if asyncapi_path:
             asyncapi_route = AsyncAPIRoute.ensure_route(asyncapi_path)
             handler = asyncapi_route(self.schema)
-            handler.set_logger(logger)  # type: ignore[attr-defined]
+            handler.set_logger(logger)
             self.routes.append((asyncapi_route.path, handler))
 
+            if asyncapi_route.try_it_out and self.broker is not None:
+                try_it_out_route = make_try_it_out_handler(
+                    self.broker,
+                    include_in_schema=asyncapi_route.include_in_schema,
+                )
+
+                try_it_out_route.update_fd_config(self.config)
+                try_it_out_route.set_logger(logger)
+
+                self.routes.append((
+                    asyncapi_route.try_it_out_url,
+                    try_it_out_route,
+                ))
+
         self._server = OuterRunState()
 
         self._log_level: int = logging.INFO

faststream/asgi/factories/__init__.py

@@ -0,0 +1,15 @@
+from .asyncapi import (
+    AsyncAPIRoute,
+    TryItOutProcessor,
+    make_asyncapi_asgi,
+    make_try_it_out_handler,
+)
+from .ping import make_ping_asgi
+
+__all__ = (
+    "AsyncAPIRoute",
+    "TryItOutProcessor",
+    "make_asyncapi_asgi",
+    "make_ping_asgi",
+    "make_try_it_out_handler",
+)

faststream/asgi/factories/asyncapi/__init__.py

@@ -0,0 +1,10 @@
+from .docs import make_asyncapi_asgi
+from .route import AsyncAPIRoute
+from .try_it_out import TryItOutProcessor, make_try_it_out_handler
+
+__all__ = (
+    "AsyncAPIRoute",
+    "TryItOutProcessor",
+    "make_asyncapi_asgi",
+    "make_try_it_out_handler",
+)