FastAPI Integration

FastAPI Integration

Ready-made lifespan context managers and Depends providers for wiring AsyncSteamClient / AsyncSteamPool into a FastAPI app.

New in pysteam-client 1.6. FastAPI is imported lazily — installing steam.aio doesn't force FastAPI into your dependency graph.

Single client

from contextlib import asynccontextmanager
from fastapi import Depends, FastAPI

from steam.aio import AsyncSteamClient
from steam.aio.integrations.fastapi import (
    get_steam_client, steam_client_lifespan,
)

client = AsyncSteamClient()

async def _login(c: AsyncSteamClient) -> None:
    await c.anonymous_login()

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with steam_client_lifespan(app, client, on_start=_login):
        yield

app = FastAPI(lifespan=lifespan)

@app.get("/product/{app_id}")
async def product(
    app_id: int,
    steam: AsyncSteamClient = Depends(get_steam_client),
):
    return await steam.get_product_info(apps=[app_id])

@app.get("/health")
async def health(steam: AsyncSteamClient = Depends(get_steam_client)):
    return steam.status.__dict__

What steam_client_lifespan does

1. await client.start() — bring up the runner thread + sync client.
2. await on_start(client) if you supplied one — the natural place to log in. A failure here propagates out and prevents the app from coming up. This is on purpose: a Steam-dependent service should not accept traffic if it can't reach Steam.
3. Attach client to app.state.steam (attribute name configurable via state_attr=).
4. yield control to FastAPI's lifespan.
5. On shutdown: clear the state attribute, await client.close().

get_steam_client

A plain sync callable — reads app.state.steam from the incoming Request. If nothing's there, it raises RuntimeError with a message telling you to wire the lifespan.

Wrap with Depends(get_steam_client) in your handler signature. FastAPI is happy to inject a sync dependency into an async handler.

Pool

Same shape, different context manager + Depends:

from steam.aio import AsyncSteamPool, PoolMember
from steam.aio.integrations.fastapi import (
    get_steam_pool, steam_pool_lifespan,
)

async def _anon(c):
    await c.anonymous_login()

pool = AsyncSteamPool([
    PoolMember(account_id="a", login=_anon),
    PoolMember(account_id="b", login=_anon),
])

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with steam_pool_lifespan(app, pool):
        yield

@app.get("/product/{app_id}")
async def product(
    app_id: int,
    pool: AsyncSteamPool = Depends(get_steam_pool),
):
    return await pool.round_robin().get_product_info(apps=[app_id])

Note the pool has no on_start parameter — each member carries its own login callable (see PoolMember).

Custom state attribute

Both steam_client_lifespan and get_steam_client accept state_attr= for cases where you want to host multiple clients on the same app:

async with steam_client_lifespan(app, alice, on_start=_login_alice, state_attr="steam_alice"):
    async with steam_client_lifespan(app, bob, on_start=_login_bob, state_attr="steam_bob"):
        yield

@app.get("/alice")
async def alice_endpoint(
    steam=Depends(lambda r: get_steam_client(r, state_attr="steam_alice")),
):
    ...

If you find yourself doing that a lot, use AsyncSteamPool instead — it handles multi-account lifecycle in one object.

Cancellation on client disconnect

FastAPI cancels the handler coroutine when the HTTP client disconnects mid-request. Because AsyncSteamClient kills the underlying gevent greenlet on cancellation, a client that hangs up while waiting for get_product_info doesn't leave a zombie greenlet on the runner thread.

Health / readiness patterns

@app.get("/health")
async def health(steam: AsyncSteamClient = Depends(get_steam_client)):
    """Liveness — the process is running."""
    return {"ok": True}

@app.get("/ready")
async def ready(steam: AsyncSteamClient = Depends(get_steam_client)):
    """Readiness — we can actually serve Steam requests."""
    s = steam.status
    if not (s.connected and s.logged_on):
        raise HTTPException(status_code=503, detail=s.__dict__)
    return s.__dict__

Pair with a Kubernetes probe:

livenessProbe:
  httpGet: { path: /health, port: 8000 }
readinessProbe:
  httpGet: { path: /ready, port: 8000 }
  periodSeconds: 5

Related pages

• AsyncSteamClient — the underlying client
• Pool — multi-account variant
• TaskIQ-Integration — same story for TaskIQ workers
• MCP — expose the client as MCP tools an LLM agent can call