dataflow.api.game_api

Generic game API with WebSocket support and Supabase integration.

This module provides a FastAPI implementation for any agent-based game in the Haive framework, with support for:

• REST endpoints for game state management

• WebSocket connections for real-time updates

• Supabase persistence for cloud storage

• Row-Level Security (RLS) for data isolation

• Multi-game support through factory patterns

The API supports any game that follows the standard Haive agent pattern, allowing for easy integration of new games.

Attributes

app
logger
module_path

Classes

GameAPI	Generic API for agent-based games with WebSocket support.
GameAPIFactory	Factory for creating game-specific APIs.
GameRequest	Base request model for creating a new game.
GameResponseBase	Base response model for game state.

Module Contents

class dataflow.api.game_api.GameAPI(app_name: str, agent_class: type[haive.core.engine.agent.agent.Agent], state_schema: type[haive.core.schema.state_schema.StateSchema], response_model: type[pydantic.BaseModel] | None = None, request_model: type[pydantic.BaseModel] | None = None, route_prefix: str = '/api/games', ws_route_prefix: str = '/ws/games')

Generic API for agent-based games with WebSocket support.

This class provides a complete API implementation for any game that follows the Haive agent pattern, with both REST endpoints and WebSocket connections for real-time updates.

app

The FastAPI application

Type:

FastAPI

agent_class

The agent class for the game

Type:

Type[Agent]

state_schema

The state schema for the game

Type:

Type[StateSchema]

socket_server

The WebSocket server

Type:

GameSocketServer

run(host: str = '0.0.0.0', port: int = 8000)

Run the API server.

agent_class

app

app_name

request_model

response_model

route_prefix = '/api/games'

socket_server

state_schema

ws_route_prefix = '/ws/games'

class dataflow.api.game_api.GameAPIFactory

Factory for creating game-specific APIs.

This class creates specialized API instances for different game types, with appropriate state schemas and agent classes for each game.

Examples

# Create a chess API chess_api = GameAPIFactory.create_chess_api()

# Run the server chess_api.run(port=8000)

static create_api(app_name: str, agent_class: type[haive.core.engine.agent.agent.Agent], state_schema: type[haive.core.schema.state_schema.StateSchema], response_model: type[pydantic.BaseModel] | None = None, request_model: type[pydantic.BaseModel] | None = None, route_prefix: str = '/api/games', ws_route_prefix: str = '/ws/games') → GameAPI

Create a game API for any agent and state schema.

Parameters:

• app_name – The name of the game/application

• agent_class – The agent class for the game

• state_schema – The state schema for the game

• response_model – Optional custom response model

• request_model – Optional custom request model

• route_prefix – The URL prefix for REST routes

• ws_route_prefix – The URL prefix for WebSocket routes

Returns:

A configured GameAPI instance

static create_chess_api() → GameAPI

Create a chess-specific API.

Returns:

A configured GameAPI instance for chess

static create_connect4_api() → GameAPI

Create a Connect4-specific API.

Returns:

A configured GameAPI instance for Connect4

static create_tic_tac_toe_api() → GameAPI

Create a Tic Tac Toe-specific API.

Returns:

A configured GameAPI instance for Tic Tac Toe

class dataflow.api.game_api.GameRequest(/, **data: Any)

Bases: pydantic.BaseModel

Base request model for creating a new game.

config_overrides: dict[str, Any] | None = None

persistence_type: str = 'supabase'

thread_id: str | None = None

user_id: str | None = None

class dataflow.api.game_api.GameResponseBase(/, **data: Any)

Bases: pydantic.BaseModel

Base response model for game state.

state: dict[str, Any]

thread_id: str

timestamp: datetime.datetime = None

dataflow.api.game_api.app

dataflow.api.game_api.logger

dataflow.api.game_api.module_path = b'.'