Skip to content

ASGI API

Run an Istos mesh inside FastAPI or Starlette. The ASGI server owns the HTTP port; Istos rides the lifespan (serving() with HTTP off).

Details: HTTP Gateway — co-hosting.

Istos.serving(serve_http=False) (the default under lifespan) leaves HTTP to the ASGI host. Pass serve_http=True only if you also want the embedded aiohttp surface in the same process.

Run an Istos mesh inside an ASGI app (FastAPI / Starlette).

The ASGI server owns the process and the HTTP port; Istos rides its lifespan — the Zenoh session opens on startup and closes on shutdown. Routes then reach the whole mesh in-process through the app (query_once / stream_query / publish_once), so there is one process and no sidecar.

from fastapi import FastAPI
from istos import Istos
from istos.http.asgi import lifespan

mesh = Istos(config=IstosZenohConfig(mode="client", ...))
api = FastAPI(lifespan=lifespan(mesh))

@api.get("/move")
async def move(distance: int):
    return await mesh.query_once("robot/move", distance=distance)

If you already have your own lifespan, drop the primitive in directly instead:

@asynccontextmanager
async def my_lifespan(api):
    async with mesh.serving():
        ...          # your own startup
        yield

lifespan(mesh)

An ASGI lifespan that starts and stops mesh. Pass it straight to FastAPI(lifespan=...) or Starlette(lifespan=...).

Source code in src/istos/http/asgi.py 
37
38
39
40
41
42
43
44
45
46
def lifespan(mesh: "Istos") -> Callable[[Any], Any]:
    """An ASGI lifespan that starts and stops ``mesh``. Pass it straight to
    ``FastAPI(lifespan=...)`` or ``Starlette(lifespan=...)``."""

    @contextlib.asynccontextmanager
    async def _lifespan(_app: Any) -> AsyncIterator[None]:
        async with mesh.serving():
            yield

    return _lifespan