SteamClient

SteamClient

steam.client.SteamClient is the gevent-based client for Steam's connection-manager (CM) protocol. It handles the encrypted handshake, login, message framing, and dispatch — with a rich event model on top.

SteamClient composes several mixins into one class:

• Apps — PICS product info, licenses, access tokens, depot keys, ticket flows.
• User — persona state, friends' presence, games-played, chat.
• Web — bootstraps an authenticated requests.Session from the CM login.
• UnifiedMessages — service-method (UM) call / response wiring.
• Leaderboards — read leaderboard entries as list-like objects.
• Friends — friend list synchronization + add / remove / block.
• GameServers — game-server listing through Steam.

The mixins are combined via steam.client.builtins.BuiltinBase, which SteamClient inherits from alongside CMClient. You don't instantiate the mixins directly — you use SteamClient and everything is there.

Requirements

Needs the client extra:

poetry install --extras client

That pulls in gevent, protobuf, and gevent-eventemitter.

Constructor and connection lifecycle

from steam.client import SteamClient

client = SteamClient()

SteamClient() takes no arguments. Instantiation is cheap — no network, no threads.

Connection flows are the three login methods:

Anonymous

result = client.anonymous_login()   # -> EResult
assert result == client.EResult.OK

Signature: anonymous_login()

No credentials, no 2FA. Enough for public PICS lookups and general CM messages that don't require licenses. Cannot download depot files — see FAQ.

CLI (interactive)

result = client.cli_login()          # prompts for username, password, 2FA
result = client.cli_login('bob')     # prompts only for password + 2FA

Signature: cli_login(username='', password='')

Best for scripts you run yourself. Handles email codes, 2FA codes, invalid passwords, and Steam-is-down retries with the same prompt loop.

Programmatic

result = client.login(
    username='bob',
    password='hunter2',
    two_factor_code='K6VKF',
)

Signature: login(username, password='', login_key=None, auth_code=None, two_factor_code=None, login_id=None)

The EVENT_AUTH_CODE_REQUIRED event fires when Steam wants a 2FA / email code — subscribe to it to build your own prompt UI:

@client.on(client.EVENT_AUTH_CODE_REQUIRED)
def on_auth_needed(is_2fa, code_mismatch):
    if is_2fa:
        code = input('Enter 2FA code: ')
        client.login(username, password, two_factor_code=code)
    else:
        code = input('Enter email code: ')
        client.login(username, password, auth_code=code)

Refresh-token / "remember me"

After a successful password login, Steam returns a login_key on the EVENT_NEW_LOGIN_KEY event. Persist that key and use relogin() next time — no password, no 2FA prompt:

if client.relogin_available:
    client.relogin()
else:
    client.cli_login()

Set client.set_credential_location('/path/to/dir') if you also want sentry files persisted; the client uses that directory to remember CM lists too.

Disconnect / reconnect

client.logout()       # send ClientLogOff, wait for the CM to close
client.disconnect()   # close the socket (idempotent)
client.reconnect()    # exponential-backoff reconnect

reconnect(maxdelay=30, retry=0) waits with exponential backoff up to maxdelay seconds. The backoff resets on a successful login.

Event model

SteamClient extends gevent_eventemitter.EventEmitter, so events are consumed via .on() decorators or .wait_event() / .wait_msg() blocking calls.

Well-known events

• client.EVENT_LOGGED_ON — fires once when login succeeds.
• client.EVENT_AUTH_CODE_REQUIRED — fires with (is_2fa: bool, code_mismatch: bool) when Steam wants a code.
• client.EVENT_NEW_LOGIN_KEY — fires when Steam issues a fresh login key. Save client.login_key from here.
• client.EVENT_DISCONNECTED — fires when the socket closes.
• client.EVENT_RECONNECT — fires just before a reconnect attempt.
• client.EVENT_CHANNEL_SECURED — CM handshake finished, safe to send().

Messages themselves also emit events, keyed by their EMsg enum:

from steam.enums.emsg import EMsg

@client.on(EMsg.ClientPersonaState)
def on_persona(msg):
    for friend in msg.body.friends:
        print(friend.player_name, friend.persona_state)

Service methods emit under their target_job_name (e.g. "Player.GetGameBadgeLevels#1").

wait_event / wait_msg

Block until an event fires:

resp = client.wait_event(EMsg.ClientPersonaState, timeout=5)

Signature: wait_event(event, timeout=None, raises=False) (inherited from EventEmitter).

wait_msg is wait_event for messages that return exactly one payload — it unwraps the tuple:

msg = client.wait_msg(EMsg.ClientPersonaState, timeout=5)
if msg:
    print(msg.body)

Signature: wait_msg(event, timeout=None, raises: bool = False)

Sending messages

SteamClient exposes three send flavours: fire-and-forget, job-tracked, and job-tracked-then-blocking.

send

send(message, body_params=None)

Fire-and-forget. message is a Msg or MsgProto. Optional body_params dict is merged into message.body for MsgProto instances.

send_job / send_job_and_wait

Job-flagged messages get a jobid in their header — Steam echoes it in the reply, and the client fires an event under job_<id> when the reply arrives. Use send_job() for a fire-and-listen pattern:

from steam.core.msg import MsgProto
from steam.enums.emsg import EMsg

jobid = client.send_job(MsgProto(EMsg.ClientRequestFriendData))
resp = client.wait_event(jobid, timeout=15)
if resp:
    (msg,) = resp
    print(msg.body)

Signature: send_job(message, body_params=None) -> str

For the common round-trip case, use send_job_and_wait:

resp_body = client.send_job_and_wait(
    MsgProto(EMsg.ClientGetNumberOfCurrentPlayersDP),
    {'appid': 570},
    timeout=10,
)
print(resp_body.player_count)

Signature: send_job_and_wait(message, body_params=None, timeout=None, raises=False)

Returns msg.body directly, or None on timeout.

send_um / send_um_and_wait

Service methods (also called "unified messages") are Steam's newer JSON-shaped RPC layer. Fire one with:

resp = client.send_um_and_wait('Player.GetGameBadgeLevels#1', {
    'playerid': 123456,
})

if resp.header.eresult == client.EResult.OK:
    print(resp.body)
else:
    print('Error:', resp.header.error_message)

Signature: send_um_and_wait(method_name, params=None, timeout=10, raises=False)

The #1 suffix is the method version. Look up available service methods in steam/protobufs/*_pb2.py — every service X { rpc Y(...) returns (...); } becomes callable.

Product info (PICS)

Full page: PICS — including the comparison table for when to use PICS vs. store HTTP vs. Web API.

The most-used method on SteamClient (via the Apps mixin):

get_product_info(apps=[], packages=[], meta_data_only=False, raw=False, auto_access_tokens=True, timeout=15)

Basic use:

resp = client.get_product_info(apps=[570, 730], packages=[123])

print(resp['apps'][570]['common']['name'])       # 'Dota 2'
print(resp['apps'][730]['common']['name'])       # 'Counter-Strike 2'
print(resp['packages'][123])

Return shape:

{
    'apps':     {570: {'common': {...}, 'extended': {...}, 'depots': {...}, ...},
                 730: {...}},
    'packages': {123: {...}},
}

Each entry also carries _missing_token, _change_number, _sha, _size bookkeeping fields.

• apps=[...] — list of app ids (ints) or dicts like {'appid': 570, 'access_token': ...} for licensed content.
• packages=[...] — same, but with packageid keys.
• meta_data_only=True — response omits the VDF payload. Fast when you only need the change number / sha.
• raw=True — buffers returned as raw bytes under _buffer, no VDF parsing.
• auto_access_tokens=True (default) — call get_access_tokens() first and fill them in. Turn off to bring your own tokens.
• timeout=15 — seconds to wait per response chunk. Large apps lists arrive as multiple chunks; timeout is per-chunk.

If an app is licensed but the request lacked its access token, the response entry has _missing_token=True. See the docstring example inside steam/client/builtins/apps.py for the retry-with-token pattern.

Full example: live smoke test

The full end-to-end flow from the README:

from steam.client import SteamClient

client = SteamClient()
assert client.anonymous_login()
resp = client.get_product_info(apps=[553850], timeout=15)  # Helldivers 2
print(resp['apps'][553850]['common']['name'])
client.logout()
client.disconnect()

Run against live Steam any time you need to confirm the whole CM handshake / anonymous-login / PICS-fetch path is intact.

Where to go next

• Downloading actual game files? CDNClient wraps a logged-in SteamClient for depot access.
• Need cookies for the web storefront from an authenticated SteamClient? Use client.get_web_session() from the Web mixin — or, if you don't have a CM login already, stand up a WebAuth directly.
• Curious what all the events look like? Grep steam/enums/emsg.py for the full EMsg catalogue.