dataflow.api.game_router_enhanced

Enhanced Game Discovery and WebSocket API for Haive Games.

This module provides a comprehensive game discovery and management system using haive-core’s unified discovery infrastructure. It creates WebSocket endpoints for real-time game state streaming and REST endpoints for game management.

Key Features:

• Automatic discovery of game agents using haive-core

• WebSocket-based real-time game state streaming

• REST API for game creation and management

• HTML client generation for browser-based gameplay

• Support for multiple concurrent game sessions

• Flexible agent initialization patterns

Architecture:

• Uses HaiveComponentDiscovery for finding game agents

• Creates dynamic routes for each discovered game

• Maintains active game sessions in memory

• Provides WebSocket connections for real-time updates

Examples

# Run as standalone server python game_router_enhanced.py

# Or integrate into existing FastAPI app from haive.dataflow.api.game_router_enhanced import get_router

app = FastAPI() games_router = get_router() app.include_router(games_router, prefix=”/games”)

Note

This implementation fixes the circular import issue and uses the unified discovery system from haive-core for consistency.

Functions

create_game_instance(game_type, game_id)	Create or retrieve a game instance.
create_game_router(game_type)	Create a router for a specific game type.
create_game_router_app()	Create a standalone FastAPI app for game routes.
discover_game_agents()	Discover game agents using the unified discovery system.
get_game_client_html(game_type)	Generate HTML for a specific game client.
get_game_instance(game_type, game_id)	Retrieve an existing game instance.
get_index_html()	Generate HTML for the index page with links to all games.
get_router()	Get a router with all game routes configured.
main()	Run the API server as standalone application.

Module Contents

dataflow.api.game_router_enhanced.create_game_instance(game_type, game_id)

Create or retrieve a game instance.

Parameters:

• game_type (str) – Type of game to create (e.g., ‘chess’, ‘checkers’).

• game_id (str) – Unique identifier for the game session.

Returns:

Game instance information including:

• agent: The instantiated game agent

• game_id: Unique game identifier

• game_type: Type of game

• created_at: Creation timestamp

• agent_info: Metadata about the agent

Return type:

Dict[str, Any]

Raises:

• ValueError – If game_type is not recognized.

• RuntimeError – If agent instantiation fails.

Note

Tries multiple initialization patterns to accommodate different agent implementations.

dataflow.api.game_router_enhanced.create_game_router(game_type)

Create a router for a specific game type.

Parameters:

game_type (str) – Type of game to create routes for.

Returns:

Configured router with WebSocket and REST endpoints.

Return type:

APIRouter

Raises:

ValueError – If game_type is not recognized.

Note

Creates the following endpoints: - WebSocket: /ws/{game_type}/{game_id} - POST: /{game_type}/games - Create new game

dataflow.api.game_router_enhanced.create_game_router_app()

Create a standalone FastAPI app for game routes.

Returns:

Configured FastAPI application with game routes.

Return type:

FastAPI

Note

Includes CORS middleware for browser compatibility.

dataflow.api.game_router_enhanced.discover_game_agents()

Discover game agents using the unified discovery system.

This function scans the haive-games package for agent classes and their corresponding state classes, registering them for use.

Discovery Process:

1. Uses HaiveComponentDiscovery to find all components

2. Filters for classes ending with ‘Agent’

3. Associates state classes with their agents

4. Groups by game module for complete game discovery

Side Effects:

Updates the global game_agents registry with discovered games.

Note

Skips base classes like ‘BaseAgent’, ‘GenericAgent’, etc.

Return type:

None

dataflow.api.game_router_enhanced.get_game_client_html(game_type)

Generate HTML for a specific game client.

Parameters:

game_type (str) – Type of game to generate client for.

Returns:

HTML content for the game client page.

Return type:

str

dataflow.api.game_router_enhanced.get_game_instance(game_type, game_id)

Retrieve an existing game instance.

Parameters:

• game_type (str) – Type of game.

• game_id (str) – Unique game identifier.

Returns:

Game instance if found, None otherwise.

Return type:

Optional[Dict[str, Any]]

dataflow.api.game_router_enhanced.get_index_html()

Generate HTML for the index page with links to all games.

Returns:

HTML content for the game index page.

Return type:

str

dataflow.api.game_router_enhanced.get_router()

Get a router with all game routes configured.

Returns:

Main router containing all game-specific routers and the index page.

Return type:

APIRouter

Note

This function triggers game discovery if not already done.

dataflow.api.game_router_enhanced.main()

Run the API server as standalone application.