Skip to content

simulatecraft.server.app

simulatecraft.server.app

FastAPI app: REST control + live websocket broadcast + inbound human input.

ControlBody

Bases: BaseModel

Optional extras for control commands (step count, rates, etc.).

WebsocketBroadcaster 

WebsocketBroadcaster(bus: EventBus)

Fans every outbound EventBus event out to connected websockets.

Source code in src/simulatecraft/server/app.py 
57
58
59
60
def __init__(self, bus: EventBus) -> None:
    self.bus = bus
    self.clients: set[WebSocket] = set()
    bus.subscribe(self._on_event)

SimulationServer 

SimulationServer(runner: Runner, host: str = '127.0.0.1', port: int = 8000)

Owns the uvicorn server + a background simulation task.

Usage::

runner = Runner(environment=env, agents=[...])
async with SimulationServer(runner) as server:
    await server.serve()
Source code in src/simulatecraft/server/app.py 
333
334
335
336
337
338
339
def __init__(self, runner: Runner, host: str = "127.0.0.1", port: int = 8000) -> None:
    self.runner = runner
    self.host = host
    self.port = port
    self.app = create_app(runner)
    self._server: Any = None
    self._sim_task: asyncio.Task[None] | None = None