Overview
The SteamAPIClient class provides an async interface to Steam’s Market API endpoints. It includes built-in rate limiting, session management, and automatic response parsing into Pydantic models.
Class Definition
from src.steamAPIclient import SteamAPIClient
client = SteamAPIClient(rate_limiter=rate_limiter)
Constructor
rate_limiter
RateLimiter | None
default:"None"
Optional shared RateLimiter instance. If None, creates its own. When using orchestrator, a shared instance should be passed.
Attributes
Base URL for Steam Market API: https://steamcommunity.com/market/
Rate limiter instance for API call throttling
HTTP session with 30-second timeout
Context Manager
The client supports async context manager protocol for automatic session cleanup:
async with SteamAPIClient(rate_limiter=limiter) as client:
result = await client.fetch_price_overview(
appid=730,
market_hash_name="AK-47 | Redline (Field-Tested)"
)
# Session automatically closed
close()
Manually close the aiohttp session.
Call this when done with the client if not using context manager.
API Methods
fetch_price_overview()
Fetch current price overview for a specific item.
async def fetch_price_overview(
self,
appid: int,
market_hash_name: str,
currency: int = 1,
country: str = "US",
language: str = "english"
) -> PriceOverviewData:
Steam application ID (e.g., 730 for CS2, 570 for Dota 2)
URL-encoded market hash name of the item
Currency code (1=USD, 3=EUR, 6=RUB, etc.)
Parsed response with current pricing dataWhether the API call succeeded
Current lowest listing price (formatted string)
Recent median sale price (formatted string)
Recent sales volume (formatted string)
result = await client.fetch_price_overview(
appid=730,
market_hash_name="AK-47 | Redline (Field-Tested)",
currency=1 # USD
)
print(f"Lowest: {result.lowest_price}") # "$10.50"
print(f"Median: {result.median_price}") # "$10.75"
print(f"Volume: {result.volume}") # "1,234"
fetch_orders_histogram()
Fetch order book histogram (buy/sell orders) for a specific item.
async def fetch_orders_histogram(
self,
appid: int,
item_nameid: int,
currency: int,
country: str = "US",
language: str = "english"
) -> OrdersHistogramData:
Numeric Steam item name ID (different from market_hash_name)
Currency code (1=USD, 3=EUR, etc.)
Parsed response with order book dataWhether the API call succeeded (Steam returns 1 for true)
Total number of buy orders
buy_order_table
List[OrderBookEntry] | None
List of buy order book entries
Total number of sell orders
sell_order_table
List[OrderBookEntry] | None
List of sell order book entries
Highest buy order price string
Lowest sell order price string
result = await client.fetch_orders_histogram(
appid=730,
item_nameid=176059193,
currency=1
)
print(f"Buy orders: {result.buy_order_count}")
print(f"Sell orders: {result.sell_order_count}")
print(f"Spread: {result.highest_buy_order} - {result.lowest_sell_order}")
fetch_orders_activity()
Fetch recent trade activity for a specific item.
async def fetch_orders_activity(
self,
item_nameid: int,
country: str = "US",
language: str = "english",
currency: int = 1,
two_factor: int = 0
) -> OrdersActivityData:
Numeric Steam item name ID
Currency code (1=USD, 3=EUR, etc.)
Two-factor authentication flag
Parsed response with recent trade activityWhether the API call succeeded
List of HTML strings containing trade activity
Unix timestamp of the response
parsed_activities
List[ActivityEntry] | None
Parsed activity entries (automatically populated)
ActivityEntry objects and stored in parsed_activities.
Example:
result = await client.fetch_orders_activity(
item_nameid=176059193,
currency=1
)
for activity in result.parsed_activities:
print(f"{activity.timestamp}: {activity.action} at {activity.price} {activity.currency}")
fetch_price_history()
Fetch complete historical price data for a specific item.
async def fetch_price_history(
self,
appid: int,
market_hash_name: str,
currency: int = 1,
country: str = "US",
language: str = "english"
) -> PriceHistoryData:
URL-encoded market hash name of the item
Currency code (1=USD, 3=EUR, etc.)
Parsed response with historical price dataWhether the API call succeeded
Currency prefix (e.g., ”$”)
Currency suffix (e.g., ”€”)
Array of [date_string, price_float, volume_string] arrays
sessionidsteamLoginSecurebrowseridsteamCountry
Example:
result = await client.fetch_price_history(
appid=730,
market_hash_name="AK-47 | Redline (Field-Tested)",
currency=1
)
for entry in result.prices:
date_str, price, volume = entry
print(f"{date_str}: ${price} (volume: {volume})")
Rate Limiting
All API methods automatically acquire a rate limit token before making requests:
# CRITICAL: Acquire rate limit token before making request
await self.rate_limiter.acquire_token()
Error Handling
All methods may raise:
aiohttp.ClientResponseError: HTTP errors (4xx, 5xx)aiohttp.ClientError: Network errors (timeout, DNS, connection refused)pydantic.ValidationError: Invalid response data structure
Complete Usage Example
import asyncio
from src.RateLimiter import RateLimiter
from src.steamAPIclient import SteamAPIClient
async def main():
# Create shared rate limiter
limiter = RateLimiter(max_requests=15, window_seconds=60.0)
# Use context manager for automatic cleanup
async with SteamAPIClient(rate_limiter=limiter) as client:
# Fetch price overview
overview = await client.fetch_price_overview(
appid=730,
market_hash_name="AK-47 | Redline (Field-Tested)"
)
print(f"Price: {overview.lowest_price}")
# Fetch order book
histogram = await client.fetch_orders_histogram(
appid=730,
item_nameid=176059193,
currency=1
)
print(f"Orders: {histogram.buy_order_count} buy, {histogram.sell_order_count} sell")
# Fetch recent activity
activity = await client.fetch_orders_activity(
item_nameid=176059193
)
print(f"Recent trades: {len(activity.parsed_activities)}")
# Fetch price history (requires auth)
history = await client.fetch_price_history(
appid=730,
market_hash_name="AK-47 | Redline (Field-Tested)"
)
print(f"Historical data points: {len(history.prices)}")
asyncio.run(main())