Serving the AsyncAPI Documentation#

Using CLI and http.server#

FastStream provides a command to serve the AsyncAPI documentation.

faststream docs serve basic:asyncapi

In the above command, the path is specified in the format of python_module:FastStream. Alternatively, you can also specify asyncapi.json or asyncapi.yaml to serve the AsyncAPI documentation.

faststream docs serve asyncapi.json

faststream docs serve asyncapi.yaml

After running the command, the AsyncAPI documentation will be served on port 8000, and the terminal should display the following logs.

INFO:     Started server process [2364992]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://localhost:8000 (Press CTRL+C to quit)

And you should be able to see the following page in your browser:

Built-in ASGI for FastStream Applications#

FastStream includes lightweight ASGI support that you can use to serve both your application and the AsyncAPI documentation.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

from faststream.asgi import AsgiFastStream
from faststream.kafka import KafkaBroker
from faststream.specification import AsyncAPI

broker = KafkaBroker()

@broker.subscriber('topic')
async def my_handler(msg: str) -> None:
    print(msg)

app = AsgiFastStream(
    broker,
    specification=AsyncAPI(),
    asyncapi_path="/docs/asyncapi",
)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Also, you can configure AsyncAPI route behavior using a special class:

1
2
3
4
5
6
7
8

from faststream.asgi import AsgiFastStream, AsyncAPIRoute
from faststream.specification import AsyncAPI

app = AsgiFastStream(
    ...,
    specification=AsyncAPI(),
    asyncapi_path=AsyncAPIRoute("/docs/asyncapi", include_in_schema=True),
)

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.

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

1
2
3
4
5
6
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:

1
2
3
4
5
6
7
8
9

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:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

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",
    ),
)

Integration with Different HTTP Frameworks (FastAPI Example)#

FastStream provides two robust approaches to combine your message broker documentation with any ASGI web frameworks. You can choose the method that best fits with your application architecture.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

from typing import AsyncIterator
from contextlib import asynccontextmanager

from fastapi import FastAPI
from faststream.asgi import make_asyncapi_asgi
from faststream.specification import AsyncAPI
from faststream.kafka import KafkaBroker

broker = KafkaBroker()

@broker.subscriber('topic')
async def my_handler(msg: str) -> None:
    print(msg)

@asynccontextmanager
async def broker_lifespan(app: FastAPI) -> AsyncIterator[None]:
    async with broker:
        await broker.start()
        yield

app = FastAPI(lifespan=broker_lifespan)
app.mount("/docs/asyncapi", make_asyncapi_asgi(AsyncAPI(broker)))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

from typing import AsyncIterator
from contextlib import asynccontextmanager

from fastapi import FastAPI, responses
from faststream.specification import get_asyncapi_html, AsyncAPI
from faststream.kafka import KafkaBroker

broker = KafkaBroker()

@broker.subscriber('topic')
async def my_handler(msg: str) -> None:
    print(msg)

@asynccontextmanager
async def broker_lifespan(app: FastAPI) -> AsyncIterator[None]:
    async with broker:
        await broker.start()
        yield

app = FastAPI(lifespan=broker_lifespan)

@app.get('/docs/asyncapi')
async def docs() -> responses.HTMLResponse:
    specification = AsyncAPI(broker).to_specification()
    return responses.HTMLResponse(get_asyncapi_html(specification))

After running the app, the documentation will be available at:

• OpenAPI Docs: http://localhost:8000/docs
• AsyncAPI Docs: http://localhost:8000/docs/asyncapi

Customizing AsyncAPI Documentation#

FastStream also provides query parameters to show and hide specific sections of AsyncAPI documentation.

You can use the following parameters control the visibility of relevant sections:

1. sidebar: Whether to include the sidebar. Default is true.
2. info: Whether to include the info section. Default is true.
3. servers: Whether to include the servers section. Default is true.
4. operations: Whether to include the operations section. Default is true.
5. messages: Whether to include the messages section. Default is true.
6. schemas: Whether to include the schemas section. Default is true.
7. errors: Whether to include the errors section. Default is true.
8. expandMessageExamples: Whether to expand message examples. Default is true.

For example, to hide the entire Servers section of the documentation, simply add servers=false as a query parameter, i.e., http://localhost:8000?servers=false. The resulting page would look like the image below:

Please use the above-listed query parameters to show and hide sections of the AsyncAPI documentation.