0
# Python A2S
1

2
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.
3

4
## Package Information
5

6
- **Package Name**: python-a2s
7
- **Language**: Python
8
- **Installation**: `pip install python-a2s`
9
- **Requirements**: Python >=3.9, no external dependencies
10

11
## Core Imports
12

13
```python
14
import a2s
15
```
16

17
Import specific functions:
18

19
```python
20
from a2s import info, players, rules, ainfo, aplayers, arules
21
from a2s import SourceInfo, GoldSrcInfo, Player
22
from a2s import BrokenMessageError, BufferExhaustedError
23
```
24

25
## Basic Usage
26

27
```python
28
import a2s
29

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

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

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

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

49
## Architecture
50

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

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

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

60
## Capabilities
61

62
### Server Information Queries
63

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

66
```python { .api }
67
def info(
68
    address: tuple[str, int],
69
    timeout: float = 3.0,
70
    encoding: str | None = "utf-8"
71
) -> SourceInfo[str] | SourceInfo[bytes] | GoldSrcInfo[str] | GoldSrcInfo[bytes]:
72
    """
73
    Query server information.
74
    
75
    Parameters:
76
    - address: Server address as (IP, port) tuple
77
    - timeout: Query timeout in seconds (default: 3.0)
78
    - encoding: String encoding or None for raw bytes (default: "utf-8")
79
    
80
    Returns:
81
    SourceInfo or GoldSrcInfo object containing server details
82
    """
83

84
async def ainfo(
85
    address: tuple[str, int],
86
    timeout: float = 3.0,
87
    encoding: str | None = "utf-8"
88
) -> SourceInfo[str] | SourceInfo[bytes] | GoldSrcInfo[str] | GoldSrcInfo[bytes]:
89
    """
90
    Async version of info().
91
    """
92
```
93

94
### Player Queries
95

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

98
```python { .api }
99
def players(
100
    address: tuple[str, int],
101
    timeout: float = 3.0,
102
    encoding: str | None = "utf-8"
103
) -> list[Player[str]] | list[Player[bytes]]:
104
    """
105
    Query list of players on server.
106
    
107
    Parameters:
108
    - address: Server address as (IP, port) tuple
109
    - timeout: Query timeout in seconds (default: 3.0)
110
    - encoding: String encoding or None for raw bytes (default: "utf-8")
111
    
112
    Returns:
113
    List of Player objects
114
    """
115

116
async def aplayers(
117
    address: tuple[str, int],
118
    timeout: float = 3.0,
119
    encoding: str | None = "utf-8"
120
) -> list[Player[str]] | list[Player[bytes]]:
121
    """
122
    Async version of players().
123
    """
124
```
125

126
### Server Rules Queries
127

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

130
```python { .api }
131
def rules(
132
    address: tuple[str, int],
133
    timeout: float = 3.0,
134
    encoding: str | None = "utf-8"
135
) -> dict[str, str] | dict[bytes, bytes]:
136
    """
137
    Query server rules and configuration.
138
    
139
    Parameters:
140
    - address: Server address as (IP, port) tuple
141
    - timeout: Query timeout in seconds (default: 3.0)
142
    - encoding: String encoding or None for raw bytes (default: "utf-8")
143
    
144
    Returns:
145
    Dictionary of rule names to values
146
    """
147

148
async def arules(
149
    address: tuple[str, int],
150
    timeout: float = 3.0,
151
    encoding: str | None = "utf-8"
152
) -> dict[str, str] | dict[bytes, bytes]:
153
    """
154
    Async version of rules().
155
    """
156
```
157

158
## Types
159

160
### Server Information Types
161

162
#### Source Engine Servers
163

164
```python { .api }
165
@dataclass
166
class SourceInfo(Generic[StrType]):
167
    """Information from Source engine servers (Half-Life 2, TF2, CS:GO, etc.)"""
168
    
169
    protocol: int
170
    """Protocol version used by the server"""
171
    
172
    server_name: StrType
173
    """Display name of the server"""
174
    
175
    map_name: StrType
176
    """The currently loaded map"""
177
    
178
    folder: StrType
179
    """Name of the game directory"""
180
    
181
    game: StrType
182
    """Name of the game"""
183
    
184
    app_id: int
185
    """App ID of the game required to connect"""
186
    
187
    player_count: int
188
    """Number of players currently connected"""
189
    
190
    max_players: int
191
    """Number of player slots available"""
192
    
193
    bot_count: int
194
    """Number of bots on the server"""
195
    
196
    server_type: StrType
197
    """Type of server: 'd' (dedicated), 'l' (non-dedicated), 'p' (SourceTV proxy)"""
198
    
199
    platform: StrType
200
    """Operating system: 'l' (Linux), 'w' (Windows), 'm' (macOS)"""
201
    
202
    password_protected: bool
203
    """Server requires a password to connect"""
204
    
205
    vac_enabled: bool
206
    """Server has VAC (Valve Anti-Cheat) enabled"""
207
    
208
    version: StrType
209
    """Version of the server software"""
210
    
211
    edf: int
212
    """Extra data field indicating which optional fields are present"""
213
    
214
    ping: float
215
    """Round-trip time for the request in seconds"""
216
    
217
    # Optional fields (presence indicated by edf flags):
218
    port: int | None = None
219
    """Port of the game server"""
220
    
221
    steam_id: int | None = None
222
    """Steam ID of the server"""
223
    
224
    stv_port: int | None = None
225
    """Port of the SourceTV server"""
226
    
227
    stv_name: StrType | None = None
228
    """Name of the SourceTV server"""
229
    
230
    keywords: StrType | None = None
231
    """Tags that describe the gamemode being played"""
232
    
233
    game_id: int | None = None
234
    """Game ID for games with app ID too high for 16-bit"""
235
    
236
    # Properties to check optional field presence:
237
    @property
238
    def has_port(self) -> bool:
239
        """Check if port field is present"""
240
        return bool(self.edf & 0x80)
241
    
242
    @property
243
    def has_steam_id(self) -> bool:
244
        """Check if steam_id field is present"""
245
        return bool(self.edf & 0x10)
246
    
247
    @property
248
    def has_stv(self) -> bool:
249
        """Check if SourceTV fields are present"""
250
        return bool(self.edf & 0x40)
251
    
252
    @property
253
    def has_keywords(self) -> bool:
254
        """Check if keywords field is present"""
255
        return bool(self.edf & 0x20)
256
    
257
    @property
258
    def has_game_id(self) -> bool:
259
        """Check if game_id field is present"""
260
        return bool(self.edf & 0x01)
261
```
262

263
#### GoldSource Engine Servers
264

265
```python { .api }
266
@dataclass
267
class GoldSrcInfo(Generic[StrType]):
268
    """Information from GoldSource engine servers (Half-Life 1, CS 1.6, etc.)"""
269
    
270
    address: StrType
271
    """IP Address and port of the server"""
272
    
273
    server_name: StrType
274
    """Display name of the server"""
275
    
276
    map_name: StrType
277
    """The currently loaded map"""
278
    
279
    folder: StrType
280
    """Name of the game directory"""
281
    
282
    game: StrType
283
    """Name of the game"""
284
    
285
    player_count: int
286
    """Number of players currently connected"""
287
    
288
    max_players: int
289
    """Number of player slots available"""
290
    
291
    protocol: int
292
    """Protocol version used by the server"""
293
    
294
    server_type: StrType
295
    """Type of server: 'd' (dedicated), 'l' (non-dedicated), 'p' (SourceTV proxy)"""
296
    
297
    platform: StrType
298
    """Operating system: 'l' (Linux), 'w' (Windows)"""
299
    
300
    password_protected: bool
301
    """Server requires a password to connect"""
302
    
303
    is_mod: bool
304
    """Server is running a Half-Life mod instead of the base game"""
305
    
306
    vac_enabled: bool
307
    """Server has VAC enabled"""
308
    
309
    bot_count: int
310
    """Number of bots on the server"""
311
    
312
    ping: float
313
    """Round-trip time for the request in seconds"""
314
    
315
    # Optional mod information (present if is_mod is True):
316
    mod_website: StrType | None
317
    """URL to the mod website"""
318
    
319
    mod_download: StrType | None
320
    """URL to download the mod"""
321
    
322
    mod_version: int | None
323
    """Version of the mod installed on the server"""
324
    
325
    mod_size: int | None
326
    """Size in bytes of the mod"""
327
    
328
    multiplayer_only: bool | None
329
    """Mod supports multiplayer only"""
330
    
331
    uses_custom_dll: bool | None
332
    """Mod uses a custom DLL"""
333
    
334
    @property
335
    def uses_hl_dll(self) -> bool | None:
336
        """Compatibility alias for uses_custom_dll"""
337
        return self.uses_custom_dll
338
```
339

340
### Player Information Type
341

342
```python { .api }
343
@dataclass
344
class Player(Generic[StrType]):
345
    """Information about a player on the server"""
346
    
347
    index: int
348
    """Entry index (usually 0)"""
349
    
350
    name: StrType
351
    """Name of the player"""
352
    
353
    score: int
354
    """Score of the player"""
355
    
356
    duration: float
357
    """Time the player has been connected to the server in seconds"""
358
```
359

360
### Generic Type Parameters
361

362
```python { .api }
363
StrType = TypeVar("StrType", str, bytes)
364
"""Type variable for string vs bytes encoding"""
365
```
366

367
## Exception Types
368

369
```python { .api }
370
class BrokenMessageError(Exception):
371
    """General decoding error for malformed server responses"""
372
    pass
373

374
class BufferExhaustedError(BrokenMessageError):
375
    """Raised when response data is shorter than expected"""
376
    pass
377
```
378

379
## Configuration Constants
380

381
```python { .api }
382
DEFAULT_TIMEOUT: float = 3.0
383
"""Default timeout in seconds for server queries"""
384

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

388
DEFAULT_RETRIES: int = 5
389
"""Default number of retry attempts for failed queries"""
390
```
391

392
## Error Handling
393

394
The library can raise several types of exceptions:
395

396
**Custom Exceptions:**
397
- `BrokenMessageError`: General decoding error for malformed responses
398
- `BufferExhaustedError`: Response data too short
399

400
**Standard Exceptions:**
401
- `socket.timeout`: No response (synchronous calls)
402
- `asyncio.TimeoutError`: No response (async calls) 
403
- `socket.gaierror`: Address resolution error
404
- `ConnectionRefusedError`: Target port closed
405
- `OSError`: Various networking errors like routing failure
406

407
## Usage Examples
408

409
### Async Usage
410

411
```python
412
import asyncio
413
import a2s
414

415
async def query_server():
416
    address = ("server.example.com", 27015)
417
    
418
    # Use async versions
419
    info = await a2s.ainfo(address)
420
    players = await a2s.aplayers(address)
421
    rules = await a2s.arules(address)
422
    
423
    print(f"Server: {info.server_name}")
424
    print(f"Players: {len(players)}")
425
    print(f"Rules: {len(rules)} settings")
426

427
# Run async function
428
asyncio.run(query_server())
429
```
430

431
### Error Handling
432

433
```python
434
import a2s
435
import socket
436

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

439
try:
440
    info = a2s.info(address, timeout=5.0)
441
    print(f"Server: {info.server_name}")
442
except a2s.BrokenMessageError:
443
    print("Server sent malformed response")
444
except socket.timeout:
445
    print("Server did not respond within timeout")
446
except ConnectionRefusedError:
447
    print("Server port is closed")
448
except OSError as e:
449
    print(f"Network error: {e}")
450
```
451

452
### Raw Bytes Mode
453

454
```python
455
import a2s
456

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

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

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