or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

index.md
tile.json

tessl/pypi-python-a2s

Library to query Source and GoldSource game servers using Valve's Server Query Protocol.

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/python-a2s@1.4.x

To install, run

npx @tessl/cli install tessl/pypi-python-a2s@1.4.0

index.mddocs/

Python A2S

Library to query Source and GoldSource game servers using Valve's Server Query Protocol. Enables developers to retrieve server information, player details, and server configuration from game servers including Half-Life, Team Fortress 2, Counter-Strike, and other Source engine games.

Package Information

  • Package Name: python-a2s
  • Language: Python
  • Installation: pip install python-a2s
  • Requirements: Python >=3.9, no external dependencies

Core Imports

import a2s

Import specific functions:

from a2s import info, players, rules, ainfo, aplayers, arules
from a2s import SourceInfo, GoldSrcInfo, Player
from a2s import BrokenMessageError, BufferExhaustedError

Basic Usage

import a2s

# Server address (IP and port)
address = ("chi-1.us.uncletopia.com", 27015)

# Query server information
server_info = a2s.info(address)
print(f"Server: {server_info.server_name}")
print(f"Map: {server_info.map_name}")
print(f"Players: {server_info.player_count}/{server_info.max_players}")

# Query current players
player_list = a2s.players(address)
for player in player_list:
    print(f"Player: {player.name}, Score: {player.score}")

# Query server rules/configuration
server_rules = a2s.rules(address)
print(f"Game mode: {server_rules.get('deathmatch', 'unknown')}")

Architecture

The library implements Valve's Server Query Protocol with a clean separation between protocol handling and transport mechanisms:

  • Protocol Layer: Dedicated protocol classes (InfoProtocol, PlayersProtocol, RulesProtocol) handle message serialization/deserialization for each query type
  • Transport Layer: Separate sync (a2s_sync) and async (a2s_async) implementations handle UDP communication and response management
  • Data Models: Strongly typed dataclasses (SourceInfo, GoldSrcInfo, Player) represent server responses with engine-specific variations
  • Error Handling: Custom exception hierarchy for protocol-level errors with fallback to standard networking exceptions

This design enables both synchronous and asynchronous usage patterns while maintaining type safety and providing consistent error handling across different game server engines.

Capabilities

Server Information Queries

Query server details including name, map, player count, game type, and various server properties. Returns different data structures for Source and GoldSource engines.

def info(
    address: tuple[str, int],
    timeout: float = 3.0,
    encoding: str | None = "utf-8"
) -> SourceInfo[str] | SourceInfo[bytes] | GoldSrcInfo[str] | GoldSrcInfo[bytes]:
    """
    Query server information.
    
    Parameters:
    - address: Server address as (IP, port) tuple
    - timeout: Query timeout in seconds (default: 3.0)
    - encoding: String encoding or None for raw bytes (default: "utf-8")
    
    Returns:
    SourceInfo or GoldSrcInfo object containing server details
    """

async def ainfo(
    address: tuple[str, int],
    timeout: float = 3.0,
    encoding: str | None = "utf-8"
) -> SourceInfo[str] | SourceInfo[bytes] | GoldSrcInfo[str] | GoldSrcInfo[bytes]:
    """
    Async version of info().
    """

Player Queries

Retrieve list of players currently connected to the server with their names, scores, and connection durations.

def players(
    address: tuple[str, int],
    timeout: float = 3.0,
    encoding: str | None = "utf-8"
) -> list[Player[str]] | list[Player[bytes]]:
    """
    Query list of players on server.
    
    Parameters:
    - address: Server address as (IP, port) tuple
    - timeout: Query timeout in seconds (default: 3.0)
    - encoding: String encoding or None for raw bytes (default: "utf-8")
    
    Returns:
    List of Player objects
    """

async def aplayers(
    address: tuple[str, int],
    timeout: float = 3.0,
    encoding: str | None = "utf-8"
) -> list[Player[str]] | list[Player[bytes]]:
    """
    Async version of players().
    """

Server Rules Queries

Retrieve server configuration settings and rules as key-value pairs.

def rules(
    address: tuple[str, int],
    timeout: float = 3.0,
    encoding: str | None = "utf-8"
) -> dict[str, str] | dict[bytes, bytes]:
    """
    Query server rules and configuration.
    
    Parameters:
    - address: Server address as (IP, port) tuple
    - timeout: Query timeout in seconds (default: 3.0)
    - encoding: String encoding or None for raw bytes (default: "utf-8")
    
    Returns:
    Dictionary of rule names to values
    """

async def arules(
    address: tuple[str, int],
    timeout: float = 3.0,
    encoding: str | None = "utf-8"
) -> dict[str, str] | dict[bytes, bytes]:
    """
    Async version of rules().
    """

Types

Server Information Types

Source Engine Servers

@dataclass
class SourceInfo(Generic[StrType]):
    """Information from Source engine servers (Half-Life 2, TF2, CS:GO, etc.)"""
    
    protocol: int
    """Protocol version used by the server"""
    
    server_name: StrType
    """Display name of the server"""
    
    map_name: StrType
    """The currently loaded map"""
    
    folder: StrType
    """Name of the game directory"""
    
    game: StrType
    """Name of the game"""
    
    app_id: int
    """App ID of the game required to connect"""
    
    player_count: int
    """Number of players currently connected"""
    
    max_players: int
    """Number of player slots available"""
    
    bot_count: int
    """Number of bots on the server"""
    
    server_type: StrType
    """Type of server: 'd' (dedicated), 'l' (non-dedicated), 'p' (SourceTV proxy)"""
    
    platform: StrType
    """Operating system: 'l' (Linux), 'w' (Windows), 'm' (macOS)"""
    
    password_protected: bool
    """Server requires a password to connect"""
    
    vac_enabled: bool
    """Server has VAC (Valve Anti-Cheat) enabled"""
    
    version: StrType
    """Version of the server software"""
    
    edf: int
    """Extra data field indicating which optional fields are present"""
    
    ping: float
    """Round-trip time for the request in seconds"""
    
    # Optional fields (presence indicated by edf flags):
    port: int | None = None
    """Port of the game server"""
    
    steam_id: int | None = None
    """Steam ID of the server"""
    
    stv_port: int | None = None
    """Port of the SourceTV server"""
    
    stv_name: StrType | None = None
    """Name of the SourceTV server"""
    
    keywords: StrType | None = None
    """Tags that describe the gamemode being played"""
    
    game_id: int | None = None
    """Game ID for games with app ID too high for 16-bit"""
    
    # Properties to check optional field presence:
    @property
    def has_port(self) -> bool:
        """Check if port field is present"""
        return bool(self.edf & 0x80)
    
    @property
    def has_steam_id(self) -> bool:
        """Check if steam_id field is present"""
        return bool(self.edf & 0x10)
    
    @property
    def has_stv(self) -> bool:
        """Check if SourceTV fields are present"""
        return bool(self.edf & 0x40)
    
    @property
    def has_keywords(self) -> bool:
        """Check if keywords field is present"""
        return bool(self.edf & 0x20)
    
    @property
    def has_game_id(self) -> bool:
        """Check if game_id field is present"""
        return bool(self.edf & 0x01)

GoldSource Engine Servers

@dataclass
class GoldSrcInfo(Generic[StrType]):
    """Information from GoldSource engine servers (Half-Life 1, CS 1.6, etc.)"""
    
    address: StrType
    """IP Address and port of the server"""
    
    server_name: StrType
    """Display name of the server"""
    
    map_name: StrType
    """The currently loaded map"""
    
    folder: StrType
    """Name of the game directory"""
    
    game: StrType
    """Name of the game"""
    
    player_count: int
    """Number of players currently connected"""
    
    max_players: int
    """Number of player slots available"""
    
    protocol: int
    """Protocol version used by the server"""
    
    server_type: StrType
    """Type of server: 'd' (dedicated), 'l' (non-dedicated), 'p' (SourceTV proxy)"""
    
    platform: StrType
    """Operating system: 'l' (Linux), 'w' (Windows)"""
    
    password_protected: bool
    """Server requires a password to connect"""
    
    is_mod: bool
    """Server is running a Half-Life mod instead of the base game"""
    
    vac_enabled: bool
    """Server has VAC enabled"""
    
    bot_count: int
    """Number of bots on the server"""
    
    ping: float
    """Round-trip time for the request in seconds"""
    
    # Optional mod information (present if is_mod is True):
    mod_website: StrType | None
    """URL to the mod website"""
    
    mod_download: StrType | None
    """URL to download the mod"""
    
    mod_version: int | None
    """Version of the mod installed on the server"""
    
    mod_size: int | None
    """Size in bytes of the mod"""
    
    multiplayer_only: bool | None
    """Mod supports multiplayer only"""
    
    uses_custom_dll: bool | None
    """Mod uses a custom DLL"""
    
    @property
    def uses_hl_dll(self) -> bool | None:
        """Compatibility alias for uses_custom_dll"""
        return self.uses_custom_dll

Player Information Type

@dataclass
class Player(Generic[StrType]):
    """Information about a player on the server"""
    
    index: int
    """Entry index (usually 0)"""
    
    name: StrType
    """Name of the player"""
    
    score: int
    """Score of the player"""
    
    duration: float
    """Time the player has been connected to the server in seconds"""

Generic Type Parameters

StrType = TypeVar("StrType", str, bytes)
"""Type variable for string vs bytes encoding"""

Exception Types

class BrokenMessageError(Exception):
    """General decoding error for malformed server responses"""
    pass

class BufferExhaustedError(BrokenMessageError):
    """Raised when response data is shorter than expected"""
    pass

Configuration Constants

DEFAULT_TIMEOUT: float = 3.0
"""Default timeout in seconds for server queries"""

DEFAULT_ENCODING: str = "utf-8"
"""Default string encoding for server responses"""

DEFAULT_RETRIES: int = 5
"""Default number of retry attempts for failed queries"""

Error Handling

The library can raise several types of exceptions:

Custom Exceptions:

  • BrokenMessageError: General decoding error for malformed responses
  • BufferExhaustedError: Response data too short

Standard Exceptions:

  • socket.timeout: No response (synchronous calls)
  • asyncio.TimeoutError: No response (async calls)
  • socket.gaierror: Address resolution error
  • ConnectionRefusedError: Target port closed
  • OSError: Various networking errors like routing failure

Usage Examples

Async Usage

import asyncio
import a2s

async def query_server():
    address = ("server.example.com", 27015)
    
    # Use async versions
    info = await a2s.ainfo(address)
    players = await a2s.aplayers(address)
    rules = await a2s.arules(address)
    
    print(f"Server: {info.server_name}")
    print(f"Players: {len(players)}")
    print(f"Rules: {len(rules)} settings")

# Run async function
asyncio.run(query_server())

Error Handling

import a2s
import socket

address = ("server.example.com", 27015)

try:
    info = a2s.info(address, timeout=5.0)
    print(f"Server: {info.server_name}")
except a2s.BrokenMessageError:
    print("Server sent malformed response")
except socket.timeout:
    print("Server did not respond within timeout")
except ConnectionRefusedError:
    print("Server port is closed")
except OSError as e:
    print(f"Network error: {e}")

Raw Bytes Mode

import a2s

address = ("server.example.com", 27015)

# Get raw bytes instead of decoded strings
info = a2s.info(address, encoding=None)
print(f"Server name (bytes): {info.server_name}")  # bytes object

# Player names as bytes
players = a2s.players(address, encoding=None)
for player in players:
    print(f"Player (bytes): {player.name}")  # bytes object