AsyncSteamClient

AsyncSteamClient

An asyncio facade around the gevent-based SteamClient. Use it from FastAPI, Starlette, TaskIQ, or any other asyncio-based app.

New in pysteam-client 1.6. Requires no changes to your existing sync code — this is additive.

Why not just use SteamClient directly?

SteamClient is built on gevent. Its recv loop, heartbeat, and per-message dispatch are all greenlets, and every blocking call (login, get_product_info, …) is a cooperative wait on gevent.socket.

Dropping that into an asyncio process is unsafe:

• steam.monkey.patch_minimal() patches stdlib sockets to greenlets → fights uvicorn's event loop, breaks httpx, motor, asyncpg, etc.
• Even without monkey-patching, SteamClient.login() blocks the calling thread until Steam replies → freezes your entire FastAPI worker.

AsyncSteamClient keeps gevent fully contained to a single background daemon thread and exposes the exact call-and-return methods a FastAPI handler actually needs, translated into async.

Design in one paragraph

Each AsyncSteamClient owns one background thread with its own isolated gevent Hub. A libev async watcher (hub.loop.async_()) is the cross-thread wake-up. Work is submitted by pushing (callable, concurrent.futures.Future) onto a stdlib queue.Queue and firing the watcher; the hub wakes, gevent.spawns the callable in its own greenlet, and the greenlet writes the outcome into the future. The asyncio side does await asyncio.wrap_future(fut) — the loop is unblocked when the future resolves from the gevent thread. No monkey-patching in the parent process.

Why not run_in_executor(None, sync_call)? Because greenlets are cooperative — they only run when someone yields on the runner thread. If we handed the thread back to a ThreadPoolExecutor between calls, the background heartbeat greenlet would stop running and the CM would kill the connection at 30s cm_stale_seconds. The runner thread keeps the gevent hub alive continuously (its main greenlet blocks on a never-set gevent Event; the async watcher is what makes forward progress happen) so background loops keep running for the full lifetime of the client.

Quick start

from steam.aio import AsyncSteamClient

async with AsyncSteamClient() as client:
    await client.anonymous_login()
    info = await client.get_product_info(apps=[440])
    print(info["apps"][440]["common"]["name"])

That's it. No monkey-patching, no gevent leaking into your app.

Public API

Session lifecycle

method	notes
await client.start()	Bring up the runner thread and construct the sync client. Idempotent. Not needed if you use async with.
await client.anonymous_login(*, raise_on_error=True)	Anonymous login. Raises SteamLoginError on non-OK by default.
await client.login(username, password, *, login_key=None, auth_code=None, two_factor_code=None, login_id=None, raise_on_error=True)	Credentialed login. See SteamClient.login for the 2FA / mail-code semantics. Passwords are NOT cached.
await client.logout()	Send ClientLogOff + wait for CM ack (5s).
await client.disconnect()	Tear down socket + kill background greenlets.
await client.close()	Full teardown: disconnect + stop the runner thread. Idempotent.

RPCs

method	notes
await client.get_product_info(apps=None, packages=None, *, meta_data_only=False, raw=False, auto_access_tokens=True, timeout=15.0)	Same shape as SteamClient.get_product_info.
await client.send_um_and_wait(method_name, params=None, *, timeout=10.0, raises=False)	Send an arbitrary Unified Messages RPC. Use for anything beyond product info (auctions, workshop, chat, inventory).

Introspection

accessor	notes
client.status → ClientStatus	JSON-serialisable snapshot. See status page.
client.logged_on → bool
client.connected → bool
client.username → str | None
client.cell_id → int	CDN routing hint.
client.relogin_available → bool	Whether Steam issued a login_key we can use to skip the password on reconnect.

Event bridge

# Single-shot
name, args = await client.wait_event("logged_on", timeout=5.0)

# Streaming
async for name, args in client.events("logged_on", "disconnected"):
    handle(name, args)

wait_event raises SteamRPCTimeoutError on timeout. The streaming iterator holds a bounded internal queue (default 256 events) — on overflow the OLDEST event is dropped and a warning is logged.

Event names include the real Steam events (connected, disconnected, channel_secured, logged_on, error, reconnect, all EMsg.* names, plus friend / chat events from the builtins) as well as the three lifecycle events emitted by AsyncSteamClient itself:

• aio.reconnecting — reconnect loop started
• aio.reconnected (attempts: int) — reconnect + relogin succeeded
• aio.reconnect_failed (attempts: int, last_error: Exception \| None) — gave up

Auto-reconnect

Enabled by default. A dropped CM connection triggers a reconnect loop on the runner thread:

1. SteamClient.reconnect(maxdelay=30) — TCP-level with jittered exponential backoff.
2. On success, replay the last login: anonymous → anonymous_login(); credentialed → sync.relogin() if Steam handed out a login_key. Passwords are never cached — if login_key isn't available, the caller has to re-authenticate.
3. Emits aio.reconnected on success, aio.reconnect_failed after policy-cap attempts.

Tune via ReconnectPolicy:

from steam.aio import AsyncSteamClient, ReconnectPolicy

client = AsyncSteamClient(
    reconnect=ReconnectPolicy(
        enabled=True,
        max_delay=30,          # cap on backoff
        max_attempts=None,     # None = retry forever; use finite for dev
    ),
)

Typed errors

Every failure derives from AsyncSteamError, so callers (including MCP tool wrappers) can pattern-match on the concrete type:

exception	raised when
SteamNotStartedError	Method called before start(). Also subclasses RuntimeError.
SteamClosedError	Method called after close(). Also subclasses RuntimeError.
SteamLoginError	Login returned non-OK. Carries .eresult for pattern-match.
SteamReconnectError	Auto-reconnect gave up. Carries .attempts + .last_error.
SteamRPCTimeoutError	RPC exceeded its timeout. Carries .timeout. Also subclasses TimeoutError.

Status + metrics hook

client.status is a JSON-serialisable dataclass — safe to return from a FastAPI /health handler or expose as an MCP tool.

from steam.aio import AsyncSteamClient

client = ...  # already started + logged in

@app.get("/health")
async def health():
    s = client.status
    return {
        "healthy": s.connected and s.logged_on,
        "connected": s.connected,
        "logged_on": s.logged_on,
        "cell_id": s.cell_id,
        "reconnect_state": s.reconnect_state,
        "reconnect_attempts": s.reconnect_attempts,
        "uptime_seconds": s.uptime_seconds,
        "last_activity_at": s.last_activity_at,
    }

Wire lifecycle + RPC events to your metrics backend via metrics_hook:

from steam.aio import AsyncSteamClient, prometheus_hook

client = AsyncSteamClient(metrics_hook=prometheus_hook())

Or your own callable — signature is (event_name: str, tags: dict) -> None. Fires client.started, client.closed, cm.connected, cm.disconnected (with intentional), reconnect.started, reconnect.succeeded / reconnect.failed (with attempts), rpc.started, rpc.succeeded (with method, duration_ms), rpc.failed (with method, duration_ms, error).

A raising hook is caught and logged — a broken metrics implementation will never take down your RPC path.

Cancellation

Cancelling the awaiting coroutine (via asyncio.wait_for, task cancellation, or a FastAPI client disconnect) kills the underlying gevent greenlet — the sync work stops instead of orphaning a socket read.

try:
    result = await asyncio.wait_for(
        client.get_product_info(apps=[440]),
        timeout=2.0,
    )
except asyncio.TimeoutError:
    # The gevent greenlet handling the RPC was killed;
    # the runner thread is free to accept new work.
    ...

Concurrency model

Concurrent await calls on the same client are serialised by gevent (single hub = single greenlet-scheduler) but the asyncio loop is never blocked. Awaiting many operations concurrently is fine — they execute in gevent's fair round-robin.

One AsyncSteamClient == one CM connection. For multi-account workloads, use AsyncSteamPool.

Related pages

• Pool — multi-account fan-out
• FastAPI-Integration — lifespan + Depends helpers
• TaskIQ-Integration — broker startup hook + dependency
• MCP — expose the client as MCP tools for LLM agents