From 5c3eb14c17211a092ed2e1100702c26406eae460 Mon Sep 17 00:00:00 2001 From: Bram van den Heuvel Date: Wed, 24 Jun 2026 09:11:21 +0200 Subject: [PATCH] Create MVP RemoteAgent --- agents/__init__.py | 3 + agents/remote.py | 99 +++++++++++++++++++++++++++++ pyserver/server.py | 154 +++++++++------------------------------------ server.py | 74 +--------------------- 4 files changed, 135 insertions(+), 195 deletions(-) create mode 100644 agents/remote.py diff --git a/agents/__init__.py b/agents/__init__.py index 91fb4ed..225b0cc 100644 --- a/agents/__init__.py +++ b/agents/__init__.py @@ -10,8 +10,11 @@ from .agent import Agent from .chaos import AgentOfChaos from .mute import MuteAgent +from .remote import RemoteAgent __all__ = [ "Agent", + "AgentOfChaos", "MuteAgent", + "RemoteAgent", ] diff --git a/agents/remote.py b/agents/remote.py new file mode 100644 index 0000000..39a41ac --- /dev/null +++ b/agents/remote.py @@ -0,0 +1,99 @@ +""" + The remote agent is the agent that is accessible on the internet. + + The common implementation of the back-end is the PyClient, although the + remote agent communicates through an API so the agent is + implementation-agnostic. +""" + +from __future__ import annotations +from typing import Any, Callable + +import requests + +from .agent import Agent, Payload + +class RemoteAgent(Agent): + """ + The RemoteAgent is the agent that is remotely accessible on an external + URL through the specified protocol. + """ + + url : str + + def __init__(self, url : str, timeout : float = 1.0) -> None: + """ + Create a new instance of the remote agent. + + :param url: URL at which the agent can be accessed. + :type url: str + :param timeout: The maximum time the client is allowed to respond. + :type timeout: float + :raises requests.HTTPError: The server could not be reached. + :raises ValueError: The server returned an invalid response. + """ + self.url = url.strip("/") + self.timeout = timeout + + # Get discovery from URL + response = requests.get(self.url.strip("/") + "/", timeout=self.timeout) + response.raise_for_status() + content = response.json() + + if not isinstance(content, dict): + raise ValueError( + "Remote agent's discovery response must be a JSON object." + ) + + raw_author = content.get("author", content.get("me.noordstar.peanuts.author", "")) + raw_name = content.get("name", "") + raw_version = content.get("version", content.get("me.noordstar.peanuts.agent.version", None)) + + # Initialize agent based on given information + super().__init__( + author="" if raw_author is None else str(raw_author), + name="" if raw_name is None else str(raw_name), + version=None if raw_version is None else str(raw_version), + profile=content, + ) + + games = content.get("games") + if isinstance(games, dict): + if "tic-tac-toe" in games: + self.add_tic_tac_toe( + on_move=lambda x : self.poll("tic-tac-toe", "", x), + profile=games["tic-tac-toe"], + ) + + def get_action(self, game: str, action: str | None) -> Callable[[dict[str, Any]], dict[str, Any]] | None: + """ + Get an action function for any given game. + + :param game: The name of the game. + :type game: str + :param action: The name of the action, or None if the game only has one action. + :type action: str | None + :return: The action with which the agent would like to act, if at all. + :rtype: Callable[[dict[str, Any]], dict[str, Any]] | None + """ + return lambda x : self.poll(game=game, action=action or "", payload=x) + + def poll( + self, + game : str, + action : str, + payload : Payload + ) -> Payload: + """ + Inquire the remote agent for a response. + """ + url = "/".join([self.url, game.strip("/"), action.strip("/")]).strip("/") + + try: + response = requests.get(url, json=payload, timeout=self.timeout) + response.raise_for_status() + content = response.json() + except (requests.RequestException, requests.HTTPError, ValueError): + return {} + else: + return content if isinstance(content, dict) else {} diff --git a/pyserver/server.py b/pyserver/server.py index 40b9b26..6b8fc9f 100644 --- a/pyserver/server.py +++ b/pyserver/server.py @@ -2,58 +2,43 @@ from __future__ import annotations +from agents import Agent from collections.abc import Callable, Mapping -from typing import Any, Optional - from flask import Flask, Response, jsonify, request from functools import wraps +from typing import Any, Optional PayloadType = dict[str, Any] GameHandler = Callable[[PayloadType], PayloadType] class PyServer: - """A tiny stateless Flask app that serves discovery and game routes.""" + """ + A tiny stateless Flask app that serves discovery and routes + game behavior to an agent. + """ - def __init__(self, - name: str, - import_name: str = __name__, - profile: Optional[PayloadType] = None, - ) -> None: + def __init__(self, agent : Agent, import_name : str = __name__) -> None: """ - Create a PyServer. + Create a PyServer that serves the behavior of an Agent. - :param name: Preferred display name for discovery. - :type name: str - :param import_name: Flask import name. + :param agent: The agent that one can communicate with. + :type agent: Agent + :param import_name: Flask application import name :type import_name: str - :param profile: Additional root-level discovery metadata. - :type profile: Optional[Dict[str, Any]] - :raises ValueError: The input contains invalid information. """ - self.name = name - self.profile = dict(profile or {}) - - if "name" in self.profile: - raise ValueError( - "Root profile metadata must not define 'name'." - ) - if "games" in self.profile: - raise ValueError( - "Root profile metadata must not define 'games'." - ) - + self.agent = agent self.app = Flask(import_name) - self.__games: dict[str, PayloadType] = {} - self.__registered_routes: set[str] = set() - # Register the root self.__add_api_endpoint("", "", lambda _ : self.__discovery()) - # self.app.add_url_rule("/", - # endpoint="botman_discovery", - # view_func=self.__discovery, - # methods=["GET"] - # ) + + # Register endpoints + for name, (_, endpoints) in self.agent.registered_games.items(): + if name == "": + continue + + for route, func in endpoints.items(): + self.__add_api_endpoint(name=name, route=route, func=func) def __add_api_endpoint(self, name : str, route : str, func : GameHandler) -> None: """ @@ -67,13 +52,6 @@ class PyServer: """ url = self.__make_url(name, route) - if url in self.__registered_routes: - raise ValueError( - f"Route already registered: {url}" - ) - - self.__registered_routes.add(url) - return self.app.add_url_rule(url, endpoint="botman_" + name.replace("/", "_"), view_func=self.__func_wrapper(func), @@ -87,11 +65,17 @@ class PyServer: :return: The personal discovery information. :rtype: dict[str, Any] """ - return { - "name": self.name, - "games": dict(self.__games), - **self.profile, - } + d = dict(name=self.agent.name, author=self.agent.author) + if self.agent.version is not None: + d["version"] = self.agent.version + + d = { **self.agent.profile, **d } + + d["games"] = {} + for game, (profile, _) in self.agent.registered_games.items(): + d["games"][game] = profile + + return d def __func_wrapper(self, func : GameHandler) -> Callable[[], Response]: """ @@ -109,82 +93,6 @@ class PyServer: def __make_url(self, name : str, route : str) -> str: return "/" + "/".join([ name.strip("/"), route.strip("/") ]).strip("/") - def add_game( - self, - name: str, - profile: PayloadType, - actions: dict[str, GameHandler], - required_actions: Optional[list[str]] = None, - ) -> None: - """ - Register a stateless game with one or more action routes. - - :param name: Base route name for the game. - :type name: str - :param profile: Game-specific discovery metadata. - :type profile: dict[str, Any] - :param actions: Mapping of action subpaths to handlers. - :type actions: dict[str, Callable[[dict[str, Any]], dict[str, Any]]] - :param required_actions: Optional completeness check. If provided, all - required action names must exist in ``actions`` and map to callables. - :type required_actions: Optional[list[str]] - :raises AssertionError: Some of the required actions aren't present. - :raises ValueError: Some of the requested URL paths are already occupied. - """ - - # Verify that all required actions are present - if required_actions is not None: - missing_actions = [ action for action in required_actions if action not in actions ] - - if len(missing_actions) > 0: - raise AssertionError( - f"Missing required action handlers: {', '.join(sorted(missing_actions))}" - ) - - # Verify that there are no duplicate URLs being registered - # Even though this will automatically be checked later, checking this - # now ensures that the operation is atomic and doesn't add - # games partially. - new_routes : set[str] = set() - - for route in actions: - url = self.__make_url(name, route) - - if url in self.__registered_routes: - raise ValueError( - "Route {url} was already registered" - ) - if url in new_routes: - raise ValueError( - "Route {url} was registered twice by the same function call" - ) - - new_routes.add(url) - - # Register all actions - for route, func in actions.items(): - self.__add_api_endpoint(name=name, route=route, func=func) - - # Add profile data - self.__games[name] = profile - - def add_tic_tac_toe(self, on_move: GameHandler, profile: PayloadType = {}) -> None: - """ - Convenience registration for tic-tac-toe. - - The game is exposed at `/tic-tac-toe`. - - :param profile: The player's custom profile. - :type profile: dict[str, Any] - :param on_move: - """ - self.add_game( - name="tic-tac-toe", - profile=profile, - actions={"": on_move}, - required_actions=[""], - ) - def start(self, host: str = "127.0.0.1", port: int = 5000, debug: bool = False, **kwargs: Any) -> None: """Start the Flask development server.""" self.app.run(host=host, port=port, debug=debug, **kwargs) diff --git a/server.py b/server.py index bf95a28..9ed7be0 100644 --- a/server.py +++ b/server.py @@ -4,10 +4,9 @@ from __future__ import annotations -import random +import agents from pyserver import PyServer -from typing import Any def main() -> int: """ @@ -16,19 +15,7 @@ def main() -> int: :return: Exit code :rtype: int """ - player = PyServer( - # Customize this to whatever you'd like to call your player - name="My super smart robot player", - - # Custom information that you can use to tell people about this player - profile={}, - - # Unless you know what you're doing, don't touch this. - import_name=__name__, - ) - - # Register games! Comment out any you don't want your player to play. - player.add_tic_tac_toe(on_move=play_tic_tac_toe, profile={}) + player = PyServer(agents.RemoteAgent(url="https://bmt001.noordstar.me/")) # Start listening for games player.start( @@ -38,62 +25,5 @@ def main() -> int: return 0 -def play_tic_tac_toe(payload : dict[str, Any]) -> dict[str, Any]: - """ - Play a game of tic-tac-toe. - - You receive a payload that looks like this: - - { - "1": "X", "2": "", "3": "O", - "4": "X", "5": "O", "6": "", - "7": "", "8": "", "9": "", - "your_token": "X" - } - - And you're expected to return a response of which field you'd like to - place your piece in. For example, if you wish to place your token in - field 7, your response should look like this: - - { "move": 7 } - - The board is arranged as follows: - - 1 | 2 | 3 - ---+---+--- - 4 | 5 | 6 - ---+---+--- - 7 | 8 | 9 - - :param payload: The incoming JSON that contains the game state. - :type payload: dict[str, Any] - :return: The move you wish to take. - :rtype: dict[str, Any] - """ - - # Try printing the payload to see what it looks like! - print(payload) - - options = [ 1, 2, 3, 4, 5, 6, 7, 8, 9, ] - - # 1. Try filtering out the impossible moves! - # If an X or O was already placed at a field, remove it from the options - - # - - # 2. Try finding two in a row! If possible, you can try to place the third - # item on the board and get 3 in a row. - - # - - # 3. Perhaps you can block the opponent from getting 3 in a row? - - # - - # Now, pick any of the remaining options. - # This is just a simple implementation. Naturally, you're welcome to try - # your own logic. - return { "move": random.choice(options) } - if __name__ == "__main__": raise SystemExit(main())