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.

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, agent_class, state_schema, response_model=None, request_model=None, route_prefix='/api/games', ws_route_prefix='/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

Initialize the game API.

Parameters:

  • app_name (str) – The name of the game/application

  • agent_class (type[haive.core.engine.agent.agent.Agent]) – The agent class for the game

  • state_schema (type[haive.core.schema.state_schema.StateSchema]) – The state schema for the game

  • response_model (type[pydantic.BaseModel] | None) – Optional custom response model

  • request_model (type[pydantic.BaseModel] | None) – Optional custom request model

  • route_prefix (str) – The URL prefix for REST routes

  • ws_route_prefix (str) – The URL prefix for WebSocket routes

run(host='0.0.0.0', port=8000)¶

Run the API server.

Parameters:
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, agent_class, state_schema, response_model=None, request_model=None, route_prefix='/api/games', ws_route_prefix='/ws/games')¶

Create a game API for any agent and state schema.

Parameters:

  • app_name (str) – The name of the game/application

  • agent_class (type[haive.core.engine.agent.agent.Agent]) – The agent class for the game

  • state_schema (type[haive.core.schema.state_schema.StateSchema]) – The state schema for the game

  • response_model (type[pydantic.BaseModel] | None) – Optional custom response model

  • request_model (type[pydantic.BaseModel] | None) – Optional custom request model

  • route_prefix (str) – The URL prefix for REST routes

  • ws_route_prefix (str) – The URL prefix for WebSocket routes

Returns:

A configured GameAPI instance

Return type:

GameAPI

static create_chess_api()¶

Create a chess-specific API.

Returns:

A configured GameAPI instance for chess

Return type:

GameAPI

static create_connect4_api()¶

Create a Connect4-specific API.

Returns:

A configured GameAPI instance for Connect4

Return type:

GameAPI

static create_tic_tac_toe_api()¶

Create a Tic Tac Toe-specific API.

Returns:

A configured GameAPI instance for Tic Tac Toe

Return type:

GameAPI

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

Bases: pydantic.BaseModel

Base request model for creating a new game.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

data (Any)

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

Bases: pydantic.BaseModel

Base response model for game state.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

data (Any)