Skip to content

MCP

Jump to bottom
cbyte edited this page Jul 23, 2026 · 1 revision

MCP (Model Context Protocol)

Expose AsyncSteamClient as Model Context Protocol tools an LLM agent can call.

New in pysteam-client 1.6. MCP SDK is imported lazily — installing steam.mcp doesn't force any framework in.

Two-layer API

  • steam.mcp.tools — the raw definitions. Framework-agnostic (Pydantic input/output schemas + async handler functions). Use these directly if you're building your own MCP server on a framework the built-in adapter doesn't cover.
  • steam.mcp.server — a thin adapter for FastMCP-compatible servers. Works with the official mcp SDK (mcp.server.fastmcp.FastMCP) and the standalone fastmcp package — both expose the same @server.tool() decorator.

Quick start

from mcp.server.fastmcp import FastMCP
from steam.aio import AsyncSteamClient
from steam.mcp import register_steam_tools

server = FastMCP("Steam")
client = AsyncSteamClient()
await client.start()
await client.anonymous_login()

register_steam_tools(server, client)
# server.run() as usual (stdio, http, whatever transport you configured)

That's it. Three MCP tools are now available to any LLM connected to server:

tool what it does
steam.status Session health snapshot (connected, logged_on, cell_id, reconnect state). No network.
steam.get_product_info Fetch Steam metadata for one or more app IDs / package IDs.
steam.send_um Arbitrary Unified Messages RPC — for anything beyond product info (auctions, workshop, chat, inventory).

Tool schemas

Every tool has a Pydantic input and output model. The LLM gets a strongly-typed contract; you get validation for free.

steam.status

class SteamStatusInput(BaseModel):
    # no parameters
    pass

class SteamStatusOutput(BaseModel):
    healthy: bool  # connected AND logged_on
    connected: bool
    logged_on: bool
    username: str | None
    cell_id: int
    reconnect_state: str  # "idle" | "reconnecting" | "failed"
    reconnect_attempts: int
    last_activity_at: float | None
    uptime_seconds: float | None

Cheap — reads local state only.

steam.get_product_info

class GetProductInfoInput(BaseModel):
    apps: list[int] = []
    packages: list[int] = []
    meta_data_only: bool = False
    timeout_seconds: float = 15.0  # bounded [0.5, 60]

class GetProductInfoOutput(BaseModel):
    apps: dict[int, dict]
    packages: dict[int, dict]

Rejects empty input (ValueError) — no burning a CM call for nothing.

steam.send_um

class SendUmInput(BaseModel):
    method_name: str  # e.g. "Player.GetGameBadgeLevels#1"
    params: dict[str, Any] = {}
    timeout_seconds: float = 10.0

class SendUmOutput(BaseModel):
    ok: bool
    body: dict[str, Any]

Error surface

Tool handlers raise typed errors that map to stable MCP error codes:

AsyncSteamError subclass MCP error code
SteamNotStartedError steam_not_started
SteamClosedError steam_closed
SteamLoginError steam_login
SteamReconnectError steam_reconnect
SteamRPCTimeoutError steam_rpctimeout
Anything else internal_error

The MCP client sees {"error": {"code": "steam_login", "message": "..."}} — stable enough to pattern-match on without regex-parsing the message.

Customising the tool set

register_steam_tools accepts an optional bindings= iterable for filtering / extending:

from steam.mcp import build_steam_tool_bindings, register_steam_tools

# Drop the low-level send_um for a locked-down deployment
safe = [b for b in build_steam_tool_bindings() if b.name != "steam.send_um"]
register_steam_tools(server, client, bindings=safe)

Prefix — multiple clients on one server

If one MCP server exposes multiple Steam accounts, prefix the tools so they don't collide:

register_steam_tools(server, alice, prefix="alice.")
register_steam_tools(server, bob,   prefix="bob.")
# Tools registered: alice.steam.status, alice.steam.get_product_info,
#                   alice.steam.send_um, bob.steam.status, ...

Building your own tools on top

The raw handlers are exported for reuse. If you want a steam.family_sharing tool the built-in set doesn't have:

from pydantic import BaseModel, Field
from steam.mcp.tools import SteamToolBinding, register_steam_tools
from steam.aio import AsyncSteamClient

class FamilyInput(BaseModel):
    steam_id: int = Field(description="64-bit Steam ID.")

class FamilyOutput(BaseModel):
    apps: list[int]

async def _handler(client: AsyncSteamClient, inp: FamilyInput) -> FamilyOutput:
    resp = await client.send_um_and_wait(
        "FamilyGroups.GetFamilyGroupForUser#1",
        {"steamid": inp.steam_id},
    )
    return FamilyOutput(apps=[a["appid"] for a in resp.body.get("apps", [])])

custom = [
    SteamToolBinding(
        name="steam.family_sharing",
        description="List apps a Steam user shares with their family group.",
        input_model=FamilyInput,
        output_model=FamilyOutput,
        handler=_handler,
    ),
]

register_steam_tools(server, client, bindings=custom)

Or mix built-ins with custom:

from steam.mcp import build_steam_tool_bindings
register_steam_tools(server, client, bindings=[*build_steam_tool_bindings(), *custom])

Full FastAPI + MCP + Steam wire-up

from contextlib import asynccontextmanager
from fastapi import FastAPI
from mcp.server.fastmcp import FastMCP
from steam.aio import AsyncSteamClient
from steam.aio.integrations.fastapi import steam_client_lifespan
from steam.mcp import register_steam_tools

client = AsyncSteamClient()
mcp_server = FastMCP("Steam")

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

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

app = FastAPI(lifespan=lifespan)
# Mount MCP on the FastAPI app however your MCP SDK version prefers
# (streamable_http_manager, sse, or as a separate stdio process).

Related pages

H47R15/steam — maintained fork of ValvePython/steam. MIT licensed.

Clone this wiki locally