FastAPI Bot Service

bot2/pyproject.toml

@@ -17,6 +17,8 @@ dependencies = [
     "tensorboard>=2.14.0",
     "scipy>=1.17.0",
     "matplotlib>=3.8.0",
+    "fastapi>=0.115.0",
+    "uvicorn>=0.34.0",
 ]
 
 [build-system]

bot2/src/bot/service/__init__.py

@@ -7,6 +7,11 @@
     SpawnBotRequest,
     SpawnBotResponse,
 )
+from bot.service.bot_service import (
+    app,
+    get_bot_manager,
+    set_bot_manager,
+)
 from bot.service.websocket_bot import (
     BotRunnerProtocol,
     WebSocketBotClient,
@@ -22,4 +27,7 @@
     "BotManager",
     "SpawnBotRequest",
     "SpawnBotResponse",
+    "app",
+    "get_bot_manager",
+    "set_bot_manager",
 ]

bot2/src/bot/service/__main__.py

@@ -0,0 +1,73 @@
+"""Entry point for running the Bot Service.
+
+Usage:
+    uv run python -m bot.service
+
+Environment variables:
+    BOT_SERVICE_PORT: Port to listen on (default: 8080)
+    BOT_SERVICE_HOST: Host to bind (default: 0.0.0.0)
+    GAME_SERVER_HTTP_URL: Game server HTTP URL (default: http://localhost:4000)
+    GAME_SERVER_WS_URL: Game server WebSocket URL (default: ws://localhost:4000/ws)
+    MODEL_REGISTRY_PATH: Path to model registry (optional)
+    DEFAULT_DEVICE: Device for inference (default: cpu)
+"""
+
+import logging
+import os
+
+import uvicorn
+
+from bot.service.bot_manager import BotManager
+from bot.service.bot_service import app, set_bot_manager
+from bot.training.registry import ModelRegistry
+
+
+def main() -> None:
+    """Initialize and run the Bot Service."""
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+    )
+    logger = logging.getLogger(__name__)
+
+    # Load configuration from environment
+    port_str = os.environ.get("BOT_SERVICE_PORT", "8080")
+    try:
+        port = int(port_str)
+    except ValueError:
+        logger.error(
+            f"Invalid BOT_SERVICE_PORT value: '{port_str}'. Must be an integer."
+        )
+        raise
+    host = os.environ.get("BOT_SERVICE_HOST", "0.0.0.0")
+    http_url = os.environ.get("GAME_SERVER_HTTP_URL", "http://localhost:4000")
+    ws_url = os.environ.get("GAME_SERVER_WS_URL", "ws://localhost:4000/ws")
+    registry_path = os.environ.get("MODEL_REGISTRY_PATH")
+    device = os.environ.get("DEFAULT_DEVICE", "cpu")
+
+    # Initialize ModelRegistry if path is provided
+    registry: ModelRegistry | None = None
+    if registry_path:
+        logger.info(f"Loading model registry from {registry_path}")
+        registry = ModelRegistry(registry_path)
+    else:
+        logger.warning("MODEL_REGISTRY_PATH not set - neural network bots unavailable")
+
+    # Initialize BotManager
+    manager = BotManager(
+        registry=registry,
+        http_url=http_url,
+        ws_url=ws_url,
+        default_device=device,
+    )
+    set_bot_manager(manager)
+
+    logger.info(f"Starting Bot Service on {host}:{port}")
+    logger.info(f"Game server: HTTP={http_url}, WS={ws_url}")
+
+    # Run uvicorn server
+    uvicorn.run(app, host=host, port=port)
+
+
+if __name__ == "__main__":
+    main()

bot2/src/bot/service/bot_service.py

@@ -0,0 +1,119 @@
+"""FastAPI Bot Service for managing game bots.
+
+This module provides the REST API layer for the Bot Service, exposing endpoints
+for spawning bots, managing active bots, listing available models, and health checks.
+"""
+
+import logging
+from contextlib import asynccontextmanager
+from typing import Annotated
+
+from fastapi import Depends, FastAPI, HTTPException, status
+
+from bot.service.bot_manager import (
+    BotInfo,
+    BotManager,
+    SpawnBotRequest,
+    SpawnBotResponse,
+)
+from bot.training.registry import ModelMetadata
+
+logger = logging.getLogger(__name__)
+
+# Global bot manager instance (initialized in __main__.py)
+_bot_manager: BotManager | None = None
+
+
+def get_bot_manager() -> BotManager:
+    """Dependency that provides the BotManager instance."""
+    if _bot_manager is None:
+        raise RuntimeError("BotManager not initialized")
+    return _bot_manager
+
+
+def set_bot_manager(manager: BotManager) -> None:
+    """Set the global BotManager instance (called from __main__.py)."""
+    global _bot_manager
+    _bot_manager = manager
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    """Application lifespan handler for startup/shutdown events."""
+    yield
+    # Shutdown: cleanup bot manager
+    if _bot_manager is not None:
+        logger.info("Shutting down bot manager...")
+        await _bot_manager.shutdown()
+
+
+app = FastAPI(
+    title="Bot Service",
+    description="Service for spawning and managing game bots",
+    version="0.1.0",
+    lifespan=lifespan,
+)
+
+
+@app.get("/health")
+def health_check() -> dict[str, str]:
+    """Health check endpoint."""
+    return {"status": "healthy"}
+
+
+@app.get("/bots/models", response_model=list[ModelMetadata])
+def list_models(
+    manager: Annotated[BotManager, Depends(get_bot_manager)],
+) -> list[ModelMetadata]:
+    """List available trained models."""
+    return manager.list_models()
+
+
+@app.post("/bots/spawn", response_model=SpawnBotResponse)
+async def spawn_bot(
+    request: SpawnBotRequest,
+    manager: Annotated[BotManager, Depends(get_bot_manager)],
+) -> SpawnBotResponse:
+    """Spawn a bot to join a game room.
+
+    Returns immediately with bot_id. Bot connects in background.
+    """
+    return await manager.spawn_bot(request)
+
+
+@app.delete("/bots/{bot_id}")
+async def delete_bot(
+    bot_id: str,
+    manager: Annotated[BotManager, Depends(get_bot_manager)],
+) -> dict[str, bool]:
+    """Remove a bot from its game."""
+    success = await manager.destroy_bot(bot_id)
+    if not success:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"Bot {bot_id} not found",
+        )
+    return {"success": True}
+
+
+@app.get("/bots", response_model=list[BotInfo])
+def list_bots(
+    manager: Annotated[BotManager, Depends(get_bot_manager)],
+) -> list[BotInfo]:
+    """List all active bots."""
+    return manager.list_bots()
+
+
+@app.get("/bots/{bot_id}", response_model=BotInfo)
+def get_bot(
+    bot_id: str,
+    manager: Annotated[BotManager, Depends(get_bot_manager)],
+) -> BotInfo:
+    """Get information about a specific bot."""
+    bot = manager.get_bot(bot_id)
+    if bot is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"Bot {bot_id} not found",
+        )
+    return bot