Bram
/
Bot-Man-Toe
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Create PyClient for hosting games

main
Bram van den Heuvel 2026-06-05 16:48:18 +02:00
commit e4c40d2d97
7 changed files with 625 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored Normal file
View File

@ -0,0 +1,2 @@
.env/
__pycache__/

14
README.md Normal file
View File

@ -0,0 +1,14 @@
# Bot-Man-Toe
Bot-Man-Toe is an attempt to create a way for players to play games against
themselves, other players, or self-trained AI players.
## Technology stack
Counterintuitively, the **servers** are participants to a game. The **clients**
are programs or browsers that mediate matches between servers.
## More
- The discovery contract is documented in `spec/README.md`.
- Python client helpers live under `pyclient/`.

80
pyclient/__init__.py Normal file
View File

@ -0,0 +1,80 @@
"""Public client entry points."""
from typing import List, Optional
import requests
from .poll import ServerAgent
class PyClient:
"""Host games between discovered server agents."""
def __init__(self, hosts: List[str]) -> None:
"""
Create a PyClient.
:param hosts: URLs of servers that can participate.
:type hosts: List[str]
:raises ValueError: If no reachable servers are provided.
"""
self.agents: List[ServerAgent] = []
self.hosts: List[str] = []
for host in hosts:
agent = self.__discover_host(host)
if agent is not None:
self.agents.append(agent)
self.hosts.append(agent.url)
if len(self.hosts) <= 0:
raise ValueError(
"No valid hosts found! Check your internet connection or verify the URLs are correct."
)
def __discover_host(self, url: str) -> Optional[ServerAgent]:
try:
return ServerAgent.from_url(url)
except (requests.exceptions.RequestException, ValueError):
return None
def play_game(self, name : str, game, timeout : float = 1.0):
"""
Play a given game. Ask the registered agents to participate.
"""
states = []
current_state = game.empty()
while current_state.winner() is None:
player : int = current_state.player_to_move()
if len(self.agents) < player:
raise KeyError(
f"The client does not have access to {player} players"
)
# Ask for move
agent = self.agents[player]
payload = agent.poll(
game=current_state.action_name(),
payload=current_state.to_dict(),
timeout=timeout,
)
# Calculate move
current_state = current_state.move(payload)
# Save the results
states.append((player, payload, current_state))
return states
def verify_host(self, url: str) -> bool:
"""
Verify whether the URL seems to contain a link to a playable server.
"""
return self.__discover_host(url) is not None
__all__ = ["PyClient", "ServerAgent"]

301
pyclient/games/tic_tac_toe.py Normal file
View File

@ -0,0 +1,301 @@
from dataclasses import dataclass
from enum import Enum, auto
from typing import Any, Dict, List, Optional
class Field(Enum):
X = auto()
O = auto()
empty = auto()
def __str__(self):
"""
Convert the field to a string.
"""
match self:
case Field.X:
return "X"
case Field.O:
return "O"
case Field.empty:
return ""
@dataclass(frozen=True)
class TicTacToe:
field_1 : Field
field_2 : Field
field_3 : Field
field_4 : Field
field_5 : Field
field_6 : Field
field_7 : Field
field_8 : Field
field_9 : Field
def __write(self, index : int, value : Field) -> "TicTacToe":
"""
Create a new copy where one field has been written with any value.
:param index: The field number to replace. (1-9)
:type index: int
:param value: The field value to write.
:type value: Field
:return: A new copy where the value was written.
:rtype: TicTacToe
:raises ValueError: The field index has already been written to.
:raises KeyError: The index is an invalid number.
"""
current_value = self.to_dict().get(str(index), None)
if current_value is None:
raise KeyError(
f"Index to write to should be 1-9, not {index}"
)
elif current_value != Field.empty:
raise ValueError(
f"Field should be empty, not {current_value}"
)
match index:
case 1:
return TicTacToe(
field_1=value,
field_2=self.field_2,
field_3=self.field_3,
field_4=self.field_4,
field_5=self.field_5,
field_6=self.field_6,
field_7=self.field_7,
field_8=self.field_8,
field_9=self.field_9,
)
case 2:
return TicTacToe(
field_1=self.field_1,
field_2=value,
field_3=self.field_3,
field_4=self.field_4,
field_5=self.field_5,
field_6=self.field_6,
field_7=self.field_7,
field_8=self.field_8,
field_9=self.field_9,
)
case 3:
return TicTacToe(
field_1=self.field_1,
field_2=self.field_2,
field_3=value,
field_4=self.field_4,
field_5=self.field_5,
field_6=self.field_6,
field_7=self.field_7,
field_8=self.field_8,
field_9=self.field_9,
)
case 4:
return TicTacToe(
field_1=self.field_1,
field_2=self.field_2,
field_3=self.field_3,
field_4=value,
field_5=self.field_5,
field_6=self.field_6,
field_7=self.field_7,
field_8=self.field_8,
field_9=self.field_9,
)
case 5:
return TicTacToe(
field_1=self.field_1,
field_2=self.field_2,
field_3=self.field_3,
field_4=self.field_4,
field_5=value,
field_6=self.field_6,
field_7=self.field_7,
field_8=self.field_8,
field_9=self.field_9,
)
case 6:
return TicTacToe(
field_1=self.field_1,
field_2=self.field_2,
field_3=self.field_3,
field_4=self.field_4,
field_5=self.field_5,
field_6=value,
field_7=self.field_7,
field_8=self.field_8,
field_9=self.field_9,
)
case 7:
return TicTacToe(
field_1=self.field_1,
field_2=self.field_2,
field_3=self.field_3,
field_4=self.field_4,
field_5=self.field_5,
field_6=self.field_6,
field_7=value,
field_8=self.field_8,
field_9=self.field_9,
)
case 8:
return TicTacToe(
field_1=self.field_1,
field_2=self.field_2,
field_3=self.field_3,
field_4=self.field_4,
field_5=self.field_5,
field_6=self.field_6,
field_7=self.field_7,
field_8=value,
field_9=self.field_9,
)
case 9:
return TicTacToe(
field_1=self.field_1,
field_2=self.field_2,
field_3=self.field_3,
field_4=self.field_4,
field_5=self.field_5,
field_6=self.field_6,
field_7=self.field_7,
field_8=self.field_8,
field_9=value,
)
case _:
raise KeyError(
f"Index to write to should be 1-9, not {index}"
)
def action_name(self):
return "tic-tac-toe"
def count_o(self) -> int:
"""
Count the number of O's on the board.
"""
return list(self.to_dict().values()).count(str(Field.O))
def count_x(self) -> int:
"""
Count the number of X's on the board.
"""
return list(self.to_dict().values()).count(str(Field.X))
@classmethod
def empty(cls):
return cls(
field_1=Field.empty, field_2=Field.empty, field_3=Field.empty,
field_4=Field.empty, field_5=Field.empty, field_6=Field.empty,
field_7=Field.empty, field_8=Field.empty, field_9=Field.empty,
)
def move_default(self) -> "TicTacToe":
"""
Have a player take a "default" move. They'll take this move
whenever their response is invalid, or when they take too long
to decide, or when they're no longer accessible.
:return: New state of the board.
:rtype: TicTacToe
:raises ValueError: When no more moves exist. By this point, someone should've already won.
"""
symbol = Field.X if self.player_to_move() == 1 else Field.O
for i in range(9):
try:
out = self.__write(i+1, symbol)
except ValueError:
# Field already occupied! Move to the next option.
pass
else:
return out
else:
# All moves seem invalid!
raise ValueError(
"No legal moves exist anymore on this tic-tac-toe board."
)
def move(self, payload : Optional[Dict[str, Any]] = None) -> "TicTacToe":
"""
Have a player make a move. Based on this information, update the
game.
:param payload: Dictionary containing the player's response.
:type payload: Optional[Dict[str, Any]]
:return: New state of the board.
:rtype: TicTacToe
:raises ValueError: When no more moves exist. By this point, someone should've already won.
"""
symbol = Field.X if self.player_to_move() == 1 else Field.O
# Extract field to place symbol at
move = 1
if isinstance(payload, dict):
move = payload.get("move", 1)
if not isinstance(move, int):
move = 1
move = min(9, max(1, move))
try:
out = self.__write(move, symbol)
except ValueError:
# Invalid move! We'll try any other move.
pass
else:
return out
# The player chose an invalid move.
return self.move_default()
def player_to_move(self) -> int:
"""
Return which player needs to move.
"""
return 1 if self.count_x() <= self.count_o() else 2
def to_dict(self) -> Dict[str, Any]:
return {
"1": str(self.field_1),
"2": str(self.field_2),
"3": str(self.field_3),
"4": str(self.field_4),
"5": str(self.field_5),
"6": str(self.field_6),
"7": str(self.field_7),
"8": str(self.field_8),
"9": str(self.field_9),
}
def winner(self) -> int | None:
"""
Returns whether the board indicates that there's a winner.
:return: The winning player, or None if there's no winner yet.
:rtype: int | None
"""
win_lines = [
[ 1, 2, 3, ],
[ 4, 5, 6, ],
[ 7, 8, 9, ],
[ 1, 4, 7, ],
[ 2, 5, 8, ],
[ 3, 6, 9, ],
[ 1, 5, 9, ],
[ 3, 5, 7, ],
]
d = self.to_dict()
for player, symbol in [ ( 1, Field.X ), ( 2, Field.O ) ]:
for win_line in win_lines:
if all(d[str(w)] == symbol for w in win_line):
return player
else:
return None

156
pyclient/poll.py Normal file
View File

@ -0,0 +1,156 @@
"""Discovery and polling helpers for Bot-Man-Toe servers."""
import multiprocessing
from queue import Empty
from typing import Any, Dict, Optional, Tuple, Union
import requests
def _poll_worker(url: str, payload: Dict[str, Any], timeout: Optional[Union[float, Tuple[float, float]]], queue: multiprocessing.Queue) -> None:
"""Run a polling request in a separate process."""
try:
response = requests.get(url, json=payload, timeout=timeout)
response.raise_for_status()
content = response.json()
if isinstance(content, dict):
queue.put(("ok", content))
return
except requests.exceptions.RequestException:
queue.put(("request_error", None))
return
except ValueError:
queue.put(("invalid", None))
return
queue.put(("invalid", None))
def _poll_request(
url: str,
payload: Dict[str, Any],
request_timeout: Optional[Union[float, Tuple[float, float]]],
deadline: float,
) -> tuple[str, Optional[Dict[str, Any]]]:
"""Execute a bounded JSON request and classify the result."""
context = multiprocessing.get_context("spawn")
queue: multiprocessing.Queue = context.Queue(maxsize=1)
process = context.Process(target=_poll_worker, args=(url, payload, request_timeout, queue))
process.daemon = True
process.start()
process.join(deadline)
if process.is_alive():
process.terminate()
process.join()
queue.close()
queue.join_thread()
return ("timeout", None)
try:
status, content = queue.get(timeout=0.1)
except Empty:
queue.close()
queue.join_thread()
return ("invalid", None)
queue.close()
queue.join_thread()
return status, content
class ServerAgent:
"""Representation of a server that can host one or more games."""
def __init__(self, url: str, name: str, games: Dict[str, Dict[str, Any]]) -> None:
"""
Create a server representation.
:param url: The URL used to discover the server.
:type url: str
:param name: Name of the server's player.
:type name: str
:param games: Games the server is willing to play.
:type games: Dict[str, Dict[str, Any]]
"""
self.url = url.strip("/")
self.name = name
self.games = games
@classmethod
def from_url(
cls,
url: str,
timeout: Optional[Union[float, Tuple[float, float]]] = 10.0,
) -> "ServerAgent":
"""
Create a server agent by polling its discovery endpoint.
The root endpoint is expected to return a JSON object with an optional
``name`` field and an optional ``games`` mapping.
:param url: The URL that the server can be reached at.
:type url: str
:param timeout: Request timeout passed to ``requests.get``.
:type timeout: Optional[Union[float, Tuple[float, float]]]
:return: The server's representation.
:rtype: ServerAgent
:raises requests.exceptions.HTTPError: If the server returns a
non-success HTTP status code.
:raises requests.exceptions.RequestException: If the request fails
before a response is received.
:raises ValueError: If the response body is not a JSON object or if the
payload contains malformed discovery fields.
"""
if timeout is None:
deadline = 10.0
elif isinstance(timeout, (int, float)):
deadline = float(timeout)
else:
deadline = sum(timeout)
status, content = _poll_request(url.rstrip("/") + "/", {}, timeout, deadline)
if status == "timeout":
raise requests.exceptions.RequestException("Server discovery request timed out.")
if status == "request_error":
raise requests.exceptions.RequestException("Server discovery request failed.")
if status != "ok" or content is None:
raise ValueError("Server discovery responses must be JSON objects.")
raw_name = content.get("name", "")
name = "" if raw_name is None else str(raw_name)
games: Dict[str, Dict[str, Any]] = {}
raw_games = content.get("games", {})
if raw_games is not None:
if not isinstance(raw_games, dict):
raise ValueError("The 'games' field must be a JSON object when provided.")
for game_name, profile in raw_games.items():
if isinstance(profile, dict):
games[str(game_name)] = profile
return cls(url=url, name=name, games=games)
def poll(self, game: str, payload: Dict[str, Any], timeout: float = 1.0) -> Optional[Dict[str, Any]]:
"""
Inquire a game to make a move.
:param game: The game the ServerAgent is asked to play.
:type game: str
:param payload: The JSON payload that represents the game's state.
:type payload: Dict[str, Any]
:param timeout: Maximum number of seconds to wait for a move.
:type timeout: float
:return: The server's response, or None if the server did not respond
in time or returned an invalid response.
:rtype: Optional[Dict[str, Any]]
"""
url = f"{self.url.rstrip('/')}/{game.lstrip('/')}"
status, content = _poll_request(url, payload, timeout, timeout)
if status != "ok":
return None
return content

5
requirements.txt Normal file
View File

@ -0,0 +1,5 @@
certifi==2026.5.20
charset-normalizer==3.4.7
idna==3.18
requests==2.34.2
urllib3==2.7.0

67
spec/README.md Normal file
View File

@ -0,0 +1,67 @@
# Bot-Man-Toe Specification
This document describes how a Bot-Man-Toe server or client is supposed to
behave.
## Terminology
A **server** is a REST API server that hosts a player willing to play games.
A **client** can feed it a game's state through a request with a JSON body,
and the server may respond with an action they wish to take.
While each API request is completely stateless and memoryless, it is typical
for clients to use the server to play full games.
## Discovery
Each server has a uniquely identifying URL at which they can be addressed.
### /
At the root of the URL, the client may send a GET request. The server should
ignore the JSON body, and respond with a profile detailing which game(s) the
server is willing to play.
#### 200 Response
The server issues this response to indicate that it's a Bot-Man-Toe server.
It'll offer some details, preferences and clarifications about itself that the
client may use.
For example, they may choose a favourite color they'd like to be displayed in,
indicate their favourite type of food or restrict what kind of roles it can
play in certain games. (e.g. it can only play as white in chess)
| Field | Type | Description |
| ----- | ---- | ----------- |
| name | string | The preferred name of the server in games. Clients may use this to create scoreboards. |
| games | {string:Game} | Collection of supported games. The key is the game's identifying API endpoint route, and the value is the server's profile information for that game. |
Notes:
- Clients should treat any unrecognized top-level fields as opaque metadata.
- Clients should ignore malformed entries inside `games` rather than failing the whole discovery step, unless the root payload itself is invalid.
- Discovery requests should be made against the root endpoint with a plain `GET`; the server may ignore the request body if one is present.
Some other metadata might be provided. When providing custom information, you
are recommended to provide this information in a way that respects the Java
package namespace guidelines.
```json
{
"name": "My self-built AI!!!1!",
"games": {
"/snake": {
"snake_color": "red",
"snake_face": "silly"
},
"/tic-tac-toe": {},
"/chess": {
"preferred_color": "white"
}
},
"org.example.my-custom-data": "foobarbaz",
"org.example.some-more-data": {}
}
```