Pool

AsyncSteamPool

Multi-account connection pool for AsyncSteamClient. One client per Steam account, all brought up concurrently at startup.

New in pysteam-client 1.6.

When to use

• TaskIQ workers farming out get_product_info calls across several licensed accounts.
• MCP server exposing tools for different personas ("act as alice" / "act as bob").
• FastAPI app juggling scraper accounts to spread CM load.

Not a connection pool in the DB sense — you don't check clients out and back in. Each get returns a shared handle; concurrent await calls on the same client are serialised inside the gevent hub (see the AsyncSteamClient concurrency model).

Quick start

from steam.aio import AsyncSteamPool, PoolMember

async def _anon_login(client):
    await client.anonymous_login()

pool = AsyncSteamPool([
    PoolMember(account_id="alice", login=_anon_login),
    PoolMember(account_id="bob",   login=_anon_login),
    PoolMember(account_id="carol", login=_anon_login),
])

async with pool:
    client = pool.round_robin()
    info = await client.get_product_info(apps=[440])

The login callable is where you put per-member auth. Anonymous is trivial; a credentialed member looks like:

async def _login_alice(client):
    r = await client.login("alice", password_for_alice, two_factor_code=...)
    # SteamLoginError already raised if it wasn't OK

PoolMember(account_id="alice", login=_login_alice)

API

Construction

AsyncSteamPool(members: Iterable[PoolMember], *, metrics_hook=None)

• members — one PoolMember per Steam account. Duplicate account_ids raise ValueError. An empty pool is rejected.
• metrics_hook — default hook used for every member's AsyncSteamClient. Per-member overrides via PoolMember(..., metrics_hook=...) — useful when you want Prometheus labels to carry account_id.

Lifecycle

method	notes
await pool.start()	Spawn all members concurrently via asyncio.gather. Individual failures are recorded in status() but don't abort the pool — the pool is usable as long as ≥1 member came up. Idempotent.
await pool.close()	Close every member concurrently. Idempotent.
async with pool as pool:	Shortcut for the above.

Selection

method	notes
pool.acquire(account_id) -> AsyncSteamClient	Get by id. KeyError if unknown, RuntimeError if that member failed to start.
pool.round_robin() -> AsyncSteamClient	Get the next ready client in cyclic order. Skips failed / closed members. RuntimeError if no members are ready.
pool.ready_clients() -> list[AsyncSteamClient]	Snapshot for fan-out with asyncio.gather.

Health

method	notes
pool.status() -> list[PoolMemberStatus]	Per-member snapshot, sorted by account_id. Each entry has ready (bool), failure (str or None), client_status (ClientStatus or None).

Recovery

method	notes
await pool.replace_member(cfg)	Swap or add a member. Existing client with the same account_id is closed first. Useful for recovering after a login-config fix without restarting the pool.

Failure model

• A member that fails to start (login rejected, network dead, etc.) is left out of the ready set. pool.round_robin() skips it. pool.acquire(id) raises RuntimeError. pool.status()[i].failure carries the exception repr for debugging.
• A member whose CM connection later drops does NOT show up as "failed" here — each AsyncSteamClient runs its own auto-reconnect loop transparently. Only if the reconnect loop hits its max_attempts does the client stop trying (see client.status.reconnect_state == "failed").
• Pool startup is best-effort. If 2 of 3 members log in successfully, the pool starts and reports the third as failed. If all 3 fail, pool is still constructed (usable for pool.status() inspection) but every acquire / round_robin raises RuntimeError.

Health endpoint pattern

from fastapi import Depends, FastAPI
from steam.aio import AsyncSteamPool
from steam.aio.integrations.fastapi import get_steam_pool

@app.get("/health/steam")
async def health(pool: AsyncSteamPool = Depends(get_steam_pool)):
    statuses = pool.status()
    return {
        "ready_members": [s.account_id for s in statuses if s.ready],
        "failed_members": [
            {"account_id": s.account_id, "reason": s.failure}
            for s in statuses if not s.ready
        ],
        "members": [
            {
                "account_id": s.account_id,
                "ready": s.ready,
                "failure": s.failure,
                "connected": s.client_status.connected if s.client_status else False,
                "logged_on": s.client_status.logged_on if s.client_status else False,
                "reconnect_state": s.client_status.reconnect_state if s.client_status else None,
            }
            for s in statuses
        ],
    }

Fan-out example

Fetch product info for a batch of app IDs, spreading across all ready pool members:

async def fan_out_get_product_info(pool, app_ids: list[int]) -> dict:
    clients = pool.ready_clients()
    if not clients:
        raise RuntimeError("no ready pool members")
    # Round-robin at the app-id level.
    chunks = [[] for _ in clients]
    for i, app_id in enumerate(app_ids):
        chunks[i % len(clients)].append(app_id)
    results = await asyncio.gather(*(
        c.get_product_info(apps=chunk)
        for c, chunk in zip(clients, chunks) if chunk
    ))
    merged: dict = {"apps": {}, "packages": {}}
    for r in results:
        merged["apps"].update(r.get("apps", {}))
        merged["packages"].update(r.get("packages", {}))
    return merged

Related pages

• AsyncSteamClient — the per-member client
• FastAPI-Integration — steam_pool_lifespan + get_steam_pool
• TaskIQ-Integration — register_steam_pool