API Reference

Condensed reference for all public Python classes, protocols, and functions in Biomata Engine. Grouped by layer. For full parameter docs and usage patterns see the corresponding guide pages.

Module prefix: all paths are relative to the package root (src/). Import as from src.contracts.action import ActionResult, etc.

Contracts

Source: src/contracts/ — Protocol-based interfaces. Implementations must satisfy these structural types; no subclassing required.

Actions — `contracts/action.py`

Class / function	Kind	Key members
`Intent`	dataclass	`action: str` `target: str \| None` `parameters: dict` `reasoning: str`
`StateMutations`	dataclass	`inventory: dict[str, int] =` (item deltas, clamped at 0) `ext: dict[str, Any] =` (forwarded to state_ext.apply_mutations())
`ActionResult`	dataclass	`success: bool` `outcome_text: str` `mutations: StateMutations = StateMutations()` `side_effects: list[dict] = []` `engine_commands: list[dict] = []` Legacy: `state_mutations=` dict still accepted, auto-converted.
`ActionHandler`	Protocol	`execute(agent, intent, context) -> ActionResult`
`ActionSchema`	dataclass	`name: str` `description: str` `parameters_schema: dict =` `execution_hint: ActionHint = HYBRID` `required_capabilities: frozenset[str] = frozenset()` `example: dict \| None = None` `prompt_block() -> str` `validate_parameters(params) -> list[ActionValidationError]` Legacy aliases: `kind=`, `tags=`, `examples=`
`ActionHint`	str enum	`HOST = "host"` \| `ENGINE = "engine"` \| `HYBRID = "hybrid"` `ActionKind = ActionHint` (backwards-compat alias)
`ActionValidationError`	dataclass	`code: str` `message: str` `field: str \| None`
`parse_intent(raw: str, valid_actions?) -> Intent`	function	Parses LLM JSON output into an `Intent`. Handles markdown fences, partial JSON; falls back to `idle`.

 World — contracts/world.py
   Class Kind Key members
 
   AgentView  dataclass  id: str
name: str
inventory: dict
ext: dict =  (state_ext snapshot)  
  World  Protocol  current_tick: int (property)
metadata: dict (property)
observe(agent_id) -> dict
apply(agent_id, result) -> None
tick() -> None  
  WorldContext  Protocol  rng: random.Random
get_agent(agent_id) -> AgentView | None
get_world_data() -> dict  
  SpatialWorld  Protocol (capability)  are_adjacent(id1, id2) -> bool. Also satisfies WorldContext.  
  VisibilityWorld  Protocol (capability)  get_nearby_agents(agent_id) -> list[AgentView]  
  PlaceableWorld  Protocol (capability)  place_agent(agent_id, **kwargs) -> None  
  ExternalWorld  Protocol (extends World)  push_observation(agent_id, obs) -> None
push_metadata(metadata) -> None
collect_commands() -> list[dict]
Check with isinstance(world, ExternalWorld).  
 
 
 Brain — contracts/brain.py
   Class Kind Key members
 
   BrainContext  dataclass  tick: int
memory: str = ""
metadata: dict = 
observation_schemas: list[ObservationSchema] = []
emit: Any = None  
  Brain  Protocol  async decide(agent_view, observation, actions, context) -> Intent  
  Closeable  Protocol (@runtime_checkable)  close() -> None — optional lifecycle protocol for brains with external resources. Engine calls close() on teardown for any brain that implements this.  
  Observation  type alias  dict[str, Any] — flat dict merged from registry providers + world observation + engine-injected keys  
 
 
 Observation — contracts/observation.py
   Class Kind Key members
 
   ObservationSchema  dataclass  name: str
description: str
payload_schema: dict = 
required_capabilities: frozenset[str] = frozenset()
examples: list[dict] = []
prompt_block() -> str
Legacy: tags= alias for required_capabilities=  
  ObservationProvider  Protocol (@runtime_checkable)  collect(agent_id, capabilities, world) -> dict[str, Any]  
 
 
 ObservationRegistry — engine/obs_registry.py
   Class Kind Key members
 
   ObservationRegistry  class  register(schema, provider) -> None
schemas() -> list[ObservationSchema]
schemas_for(capabilities) -> list[ObservationSchema]
collect(agent_id, capabilities, world) -> dict[str, Any]
observations_prompt_section(capabilities?) -> str  
 
 
 Memory — contracts/memory.py
   Class Kind Key members
 
   Memory  Protocol  store(tick, observation, intent, outcome) -> None
recall(n) -> str (formatted string, not list)
serialize() -> bytes
restore(data: bytes) -> None  
 
 
 State — contracts/state.py
   Class Kind Key members
 
   AgentStateExtension  Protocol  tick() -> None
apply_mutations(mutations: dict) -> None
snapshot() -> dict
to_prompt_str() -> str
urgent_advice() -> str (return "" if nothing urgent)
serialize() -> bytes
restore(data: bytes) -> None  
 
 
 Social — contracts/social.py
   Class Kind Key members
 
   SocialSystem  Protocol  add_agent(agent_id: str, name: str) -> None
update(from_id, to_id, delta) -> None
relationship(from_id, to_id) -> float
describe(agent_id) -> str
serialize() -> bytes
restore(data: bytes) -> None  
 
 
 Snapshot — contracts/snapshot.py
   Class / function Kind Key members
 
   Snapshotable  Protocol  serialize() -> bytes
restore(data: bytes) -> None  
  AgentSnapshot  dataclass  id: str
name: str
inventory: dict
memory: bytes (from Memory.serialize())
state_ext: bytes | None
brain: bytes | None (if brain implements Snapshotable)  
  SimulationSnapshot  dataclass  version: str (= "1")
tick: int
rng_state: Any (random.Random state)
config: dict
agents: list[AgentSnapshot]
world: bytes | None
social: bytes | None
scheduler: bytes | None
created_at: str (ISO 8601 UTC)
is_complete() -> bool (world and social both captured)  
  SnapshotError  exception  Raised on version mismatch or restore failure.  
  save_to_file(snapshot, path) -> None  function  Persists SimulationSnapshot to disk via pickle.  
  load_from_file(path) -> SimulationSnapshot  function  Loads from a pickle file; raises SnapshotError on version mismatch or missing file.  
 
 
 Engine
 Source: src/engine/ — Core runtime classes. Not imported by the service or transport layers directly.
 Simulation — engine/simulation.py
   Class / item Kind Key members
 
   SimulationConfig  dataclass  ticks: int = 20
seed: int = 42
log_level: str = "normal"  
  AgentTickResult  dataclass  agent_id: str
agent_name: str
intent: Intent
result: ActionResult  
  TickSummary  dataclass  tick: int
agent_results: list[AgentTickResult]
errors: list[tuple[str, str]] (agent_id, message)
engine_commands() -> list[dict]  
  Simulation  class  __init__(agents, world, registry, bus?, scheduler?, config?, social?, obs_registry?, roles?)
async run() -> None
async run_tick() -> TickSummary
register_agent(definition) -> Agent
unregister_agent(agent_id) -> Agent
snapshot() -> SimulationSnapshot
restore(snap) -> None
save_snapshot(path) -> None
load_snapshot(path) -> None
close() -> None
__enter__() / __exit__() (context manager)
classmethod from_config(yaml_path) -> Simulation  
 
 
 Agent — engine/agent.py
   Class Kind Key members
 
   Agent  dataclass  id: str
name: str
brain: Brain
memory: Memory
inventory: dict = 
state_ext: AgentStateExtension | None = None
capabilities: frozenset[str] = frozenset()
metadata: dict =  
 
 
 AgentRuntime — engine/agent_runtime.py
   Class Kind Key members
 
   AgentRuntime  class  __init__(agent, world, registry, social?, event_bus?)
async step(tick) -> tuple[Intent | None, ActionResult | None]
_build_observation(agent_id) -> dict  
 
 
 step() pipeline: tick state ext → build observation → brain.decide() → registry.dispatch() → apply mutations → store memory → emit events.
 ActionRegistry — engine/registry.py
   Class Kind Key members
 
   ActionRegistry  class  register(schema, handler) -> None
schemas() -> list[ActionSchema]
schemas_for(capabilities: frozenset[str]) -> list[ActionSchema]
dispatch(intent, agent_view, world_context) -> ActionResult  
 
 
 Schedulers — engine/scheduler.py
   Class Kind Key members
 
   Scheduler  Protocol  async run_tick(agents, step_fn) -> list[tuple[Agent, Any]]  
  SimultaneousScheduler  class  Runs all agents concurrently via asyncio.gather. Default scheduler.  
  SequentialScheduler  class  __init__(order: list[str] | None = None)
Runs agents one by one in declared order. Deterministic.  
 
 
 EventBus — engine/event_bus.py
   Class / constant Kind Key members
 
   Event  dataclass  type: str
tick: int
agent_id: str | None
data: dict  
  EventBus  class  subscribe(event_type, handler) -> None
unsubscribe(event_type, handler) -> None
emit(event) -> None  
  Event type constants  str constants  TICK_START, TICK_END
ACTION_COMPLETED, ACTION_FAILED
BRAIN_DECIDED
AGENT_STEP_ERROR
AGENT_REGISTERED, AGENT_UNREGISTERED  
  SocialEffectSubscriber  class  Applies social side_effects from ACTION_COMPLETED events to SocialSystem.  
  EventLogSubscriber  class  Appends a formatted line to a bounded deque(maxlen=1000) for each ACTION_COMPLETED event. Access via log_sub.tail(n).  
  ObservabilitySubscriber  class  Calls user-provided hooks: on_tick, on_action, on_failure, on_event. All hooks are optional.  
 
 
 Service
 Source: src/service/ — Transport-agnostic session layer. Transport adapters import from here, never from src/engine/ directly.
 Session — service/session.py
   Class / function Kind Key members
 
   SessionError  exception  Raised when a method is called in an incompatible SessionState or TickMode.  
  SimulationSession  class  session_id: str (property)
state: SessionState (property)
tick: int (property)
tick_mode: TickMode (property)
async step(req: StepRequest | None) -> StepResponse
async run() -> None
pause() -> None
resume() -> None
shutdown() -> None
subscribe(event_type, handler) -> str (returns subscription_id)
unsubscribe(subscription_id: str) -> None
register_agent(definition) -> Agent
unregister_agent(agent_id) -> Agent
snapshot() -> SimulationSnapshot
restore(snap) -> None
status() -> SimulationStatus  
  create_session(sim, tick_mode?) -> SimulationSession  function  Factory. tick_mode defaults to TickMode.HOST_DRIVEN. Wires EventStreamAdapter to the simulation's EventBus.  
 
 
 DTOs — service/dto.py
   Class Kind Key members
 
   AgentObservationDTO  dataclass  agent_id: str
observation: dict  
  StepRequest  dataclass  agent_observations: list[AgentObservationDTO] = []
world_metadata: dict =  
  AgentDecisionDTO  dataclass  agent_id: str
agent_name: str
action: str
parameters: dict = 
outcome_text: str = ""
engine_commands: list[dict] = []
error: str | None = None  
  StepResponse  dataclass  tick: int
decisions: list[AgentDecisionDTO]
errors: list[tuple[str, str]] (agent_id, message)
engine_commands() -> list[dict]  
  ServiceEvent  dataclass  session_id: str
event_type: str
tick: int
agent_id: str
data: dict =  
  SimulationStatus  dataclass  session_id: str
state: str
tick: int
config_ticks: int
agent_count: int
has_world_snap: bool
tick_mode: str = "host_driven"  
 
 
 Interfaces — service/interfaces.py
   Class Kind Members
 
   TickMode  str enum  HOST_DRIVEN = "host_driven" | AUTONOMOUS = "autonomous"  
  SessionState  str enum  CREATED | RUNNING | PAUSED | STOPPED | ERROR  
  EventHandler  type alias  Callable[[ServiceEvent], None]  
  SimulationController  Protocol  Structural type satisfied by SimulationSession. Declares all lifecycle + step methods.  
 
 
 EventStreamAdapter — service/events.py
   Class Kind Key members
 
   EventStreamAdapter  class  subscribe(event_type, handler) -> None
unsubscribe(event_type, handler) -> None
close() -> None  
 
 
 Single wildcard listener on the engine EventBus. Fans out to per-subscription handlers bucketed by event_type. close() removes the wildcard listener.
 Transport
 Source: src/transport/websocket/ — WebSocket adapter. Never imports from src/engine/ directly.
 WebSocketServer — transport/websocket/server.py
   Class Kind Key members
 
   WebSocketServer  class  async start() -> None
async stop() -> None
async wait_for_termination() -> None
async serve(websocket) -> None
classmethod from_simulation(sim, host?, port?, tick_mode?) -> WebSocketServer
classmethod from_config(yaml_path, host?, port?, tick_mode?) -> WebSocketServer  
 
 
 Protocol constants — transport/websocket/protocol.py
   Class Members
 
   Method  HEALTH_CHECK, REGISTER_AGENT, REMOVE_AGENT, SEND_OBSERVATION, TICK, PAUSE, RESUME, SNAPSHOT, RESTORE, SUBSCRIBE_EVENTS, UNSUBSCRIBE_EVENTS  
  ErrorCode  PARSE_ERROR = -32700
INVALID_REQUEST = -32600
METHOD_NOT_FOUND = -32601
INVALID_PARAMS = -32602
INTERNAL_ERROR = -32603
VERSION_MISMATCH = -32000
SESSION_ERROR = -1
AGENT_EXISTS = -2
AGENT_NOT_FOUND = -3
IMPORT_ERROR = -4
SNAPSHOT_INVALID = -5  
  ProtocolError  Exception with code: int, name: str, message: str  
 
 
   Function Returns
 
  build_hello(session_id, tick_mode?) -> dict hlo frame — capabilities computed internally from tick_mode
 build_response(req_id, result) -> dict success res frame
 build_error(req_id, error) -> dict error res frame
 build_event(session_id, seq, service_event) -> dict evt frame
 
 
 Plugins
 Source: src/plugins/ — Built-in implementations and the external-world adapter.
 OllamaLLMBrain — plugins/builtin/ollama/brain.py
   Class Kind Key members
 
   Personality  dataclass  traits: list[str]
goals: list[str]
backstory: str  
  OllamaLLMBrain  class (Brain)  __init__(personality?, llm_config?, system_prompt?, **kwargs)
llm_config keys: model (default "qwen2.5:14b"), base_url (default "http://localhost:11434"), temperature, max_concurrent
async decide(agent, observation, actions, context) -> Intent  
 
 
 IdleBrain — plugins/builtin/idle_brain/brain.py
   Class Kind Key members
 
   IdleBrain  class (Brain)  __init__(action="idle", target=None, reasoning="...", parameters=)
Always returns the configured static Intent. Useful for testing and placeholders.  
 
 
 SimpleMemory — plugins/builtin/simple_memory/memory.py
   Class Kind Key members
 
   SimpleMemory  class (Memory)  __init__(capacity: int = 20)
Rolling deque. Oldest entries dropped at capacity.  
 
 
 ReplayBrain / JSONReplayRecorderSubscriber
   Class Module Key members
 
   ReplayBrain  plugins/builtin/replay_brain/brain.py  __init__(mode="replay", path="replay.jsonl", inner?, llm_config?, **kwargs)
mode="record": wraps an inner brain and records all intents to JSONL.
mode="replay": reads intents from a JSONL file deterministically. Implements Closeable and Snapshotable.  
  JSONReplayRecorderSubscriber  plugins/builtin/replay_recorder/recorder.py  __init__(path: str | Path)
EventBus subscriber that writes one JSON record per event to a JSONL file. Subscribe to BRAIN_DECIDED, ACTION_COMPLETED, or "*". Implements context manager; call close() when done.  
 
 
 WeightedGraphSocial — plugins/builtin/simple_social/social.py
   Class Kind Key members
 
   WeightedGraphSocial  class (SocialSystem)  Directed graph via networkx. Weights clamped to [-1.0, 1.0]. Fully serializable/restorable for snapshots.  
 
 
 HostedWorld — plugins/external/world.py
   Class Kind Key members
 
   HostedWorld  class  Implements World, WorldContext, ExternalWorld, Snapshotable via duck-typing.

Push (host → engine):
push_observation(agent_id, obs: dict) -> None
push_metadata(metadata: dict) -> None

Pull (engine → host):
collect_commands() -> list[dict]

World protocol:
observe(agent_id) -> dict
apply(agent_id, result) -> None
tick() -> None
current_tick: int
metadata: dict  
 
 
 Config
 Source: src/config/loader.py
   Function Description
 
   load_simulation(yaml_path: str) -> Simulation  Parses a YAML config file and instantiates a Simulation. Called by Simulation.from_config(). Uses importlib for dynamic class loading.  
 
 
 YAML fields recognized by the loader:
   Field Type Description
 
  engine.ticks int (default 20) Total ticks to run (autonomous mode)
 engine.seed int (default 42) RNG seed for determinism
 engine.scheduler sequential | simultaneous Agent execution order
 engine.scheduler_order list[str] Optional explicit agent_id order for sequential scheduler
 engine.log_level normal | verbose | quiet Logging verbosity
 world.class dotted path (required) World implementation to instantiate
 registry.class dotted path Callable returning a populated ActionRegistry. Optional — defaults to empty registry.
 observations.class dotted path Optional callable returning an ObservationRegistry
 social.class dotted path Optional SocialSystem implementation
 llm dict Top-level LLM config dict forwarded to brain constructors (e.g. model, base_url)
 roles dict Named role definitions — each with capabilities and brain block. Agents may reference a role via agents[].role.
 agents[].id str Unique agent identifier
 agents[].name str Display name
 agents[].role str Reference to a declared role; inherits capabilities and brain
 agents[].position dict Passed to PlaceableWorld.place_agent() if world supports it
 agents[].brain.class dotted path Brain implementation (required unless role supplies one)
 agents[].state_ext.class dotted path AgentStateExtension implementation
 agents[].capabilities list[str] Capability tags (additive union with role capabilities)
 agents[].memory.class dotted path Memory implementation (default: SimpleMemory)
 agents[].inventory dict Starting item counts
 agents[].metadata dict Arbitrary key/value data attached to agent; not consumed by tick loop
 
 
 CLI
 Source: src/cli/main.py
  bash
 # Run a simulation from a YAML config
python -m src.cli.main run <path/to/sim.yaml>

# Start WebSocket server (host-driven, default — binds 127.0.0.1)
biomata-ws --config <path/to/sim.yaml> --port 8765

# Bind on all interfaces (LAN/remote access)
biomata-ws --config <path/to/sim.yaml> --host 0.0.0.0 --port 8765

# Start WebSocket server (autonomous mode)
biomata-ws --config <path/to/sim.yaml> --port 8765 --mode autonomous
 
 biomata-ws is the entry point declared in pyproject.toml under [project.scripts]. It requires the websocket extra (pip install -e ".[websocket]").
 Unity SDK (C#)
 Full details: Unity SDK page. Quick class inventory:
   Class Namespace / file Role
 
   BiomataManager  Runtime/Unity/  MonoBehaviour singleton. Inspector-configurable connection settings. Exposes Client property and C# events.  
  SimulationClient  Runtime/Core/  Main entry point. Properties: Agents, Ticks, Observations, Events, Snapshots. ConnectAsync / DisconnectAsync.  
  BiomataConfig  Runtime/Core/  Connection config: Host, Port, UseTls, ConnectTimeoutSeconds, DefaultCallTimeoutSeconds, Retry.  
  AgentClient  Runtime/Clients/  RegisterAsync(AgentRegistration), RemoveAsync(agentId), TryRemoveAsync(agentId).  
  TickClient  Runtime/Clients/  TickAsync(observations, worldMetadata, ct), PauseAsync(ct), ResumeAsync(ct).  
  EventStreamClient  Runtime/Clients/  Subscribe / unsubscribe to server-side event streams. Raises per-type C# events.  
  SnapshotClient  Runtime/Clients/  SnapshotAsync(ct), RestoreAsync(snapshot, ct).  
  AgentRegistration  Runtime/Models/  DTO: AgentId, AgentName, BrainClass, BrainConfig, MemoryClass, MemoryConfig.  
  AgentObservationData  Runtime/Models/  DTO: AgentId, Observation (Dictionary).  
  TickResult  Runtime/Models/  Tick, Decisions (IReadOnlyList), Errors, AllCommands, ForAgent(agentId) O(1).  
  AgentDecisionResult  Runtime/Models/  Per-agent decision: AgentId, Action, Target, Reasoning, Success, Outcome, EngineCommands.  
  RetryConfig  Runtime/Core/  MaxAttempts, InitialDelaySeconds, MaxDelaySeconds, Multiplier, JitterFraction.

Class	Kind	Key members
`AgentView`	dataclass	`id: str` `name: str` `inventory: dict` `ext: dict =` (state_ext snapshot)
`World`	Protocol	`current_tick: int` (property) `metadata: dict` (property) `observe(agent_id) -> dict` `apply(agent_id, result) -> None` `tick() -> None`
`WorldContext`	Protocol	`rng: random.Random` `get_agent(agent_id) -> AgentView \| None` `get_world_data() -> dict`
`SpatialWorld`	Protocol (capability)	`are_adjacent(id1, id2) -> bool`. Also satisfies `WorldContext`.
`VisibilityWorld`	Protocol (capability)	`get_nearby_agents(agent_id) -> list[AgentView]`
`PlaceableWorld`	Protocol (capability)	`place_agent(agent_id, **kwargs) -> None`
`ExternalWorld`	Protocol (extends World)	`push_observation(agent_id, obs) -> None` `push_metadata(metadata) -> None` `collect_commands() -> list[dict]` Check with `isinstance(world, ExternalWorld)`.

Class	Kind	Key members
`BrainContext`	dataclass	`tick: int` `memory: str = ""` `metadata: dict =` `observation_schemas: list[ObservationSchema] = []` `emit: Any = None`
`Brain`	Protocol	`async decide(agent_view, observation, actions, context) -> Intent`
`Closeable`	Protocol (`@runtime_checkable`)	`close() -> None` — optional lifecycle protocol for brains with external resources. Engine calls `close()` on teardown for any brain that implements this.
`Observation`	type alias	`dict[str, Any]` — flat dict merged from registry providers + world observation + engine-injected keys

Class	Kind	Key members
`ObservationSchema`	dataclass	`name: str` `description: str` `payload_schema: dict =` `required_capabilities: frozenset[str] = frozenset()` `examples: list[dict] = []` `prompt_block() -> str` Legacy: `tags=` alias for `required_capabilities=`
`ObservationProvider`	Protocol (`@runtime_checkable`)	`collect(agent_id, capabilities, world) -> dict[str, Any]`

Class / function	Kind	Key members
`Snapshotable`	Protocol	`serialize() -> bytes` `restore(data: bytes) -> None`
`AgentSnapshot`	dataclass	`id: str` `name: str` `inventory: dict` `memory: bytes` (from `Memory.serialize()`) `state_ext: bytes \| None` `brain: bytes \| None` (if brain implements `Snapshotable`)
`SimulationSnapshot`	dataclass	`version: str` (= `"1"`) `tick: int` `rng_state: Any` (random.Random state) `config: dict` `agents: list[AgentSnapshot]` `world: bytes \| None` `social: bytes \| None` `scheduler: bytes \| None` `created_at: str` (ISO 8601 UTC) `is_complete() -> bool` (world and social both captured)
`SnapshotError`	exception	Raised on version mismatch or restore failure.
`save_to_file(snapshot, path) -> None`	function	Persists `SimulationSnapshot` to disk via pickle.
`load_from_file(path) -> SimulationSnapshot`	function	Loads from a pickle file; raises `SnapshotError` on version mismatch or missing file.

Class / item	Kind	Key members
`SimulationConfig`	dataclass	`ticks: int = 20` `seed: int = 42` `log_level: str = "normal"`
`AgentTickResult`	dataclass	`agent_id: str` `agent_name: str` `intent: Intent` `result: ActionResult`
`TickSummary`	dataclass	`tick: int` `agent_results: list[AgentTickResult]` `errors: list[tuple[str, str]]` (agent_id, message) `engine_commands() -> list[dict]`
`Simulation`	class	`__init__(agents, world, registry, bus?, scheduler?, config?, social?, obs_registry?, roles?)` `async run() -> None` `async run_tick() -> TickSummary` `register_agent(definition) -> Agent` `unregister_agent(agent_id) -> Agent` `snapshot() -> SimulationSnapshot` `restore(snap) -> None` `save_snapshot(path) -> None` `load_snapshot(path) -> None` `close() -> None` `__enter__() / __exit__()` (context manager) `classmethod from_config(yaml_path) -> Simulation`

API Reference

Contracts

Actions — `contracts/action.py`

World — `contracts/world.py`

Brain — `contracts/brain.py`

Observation — `contracts/observation.py`

ObservationRegistry — `engine/obs_registry.py`

Memory — `contracts/memory.py`

State — `contracts/state.py`

Snapshot — `contracts/snapshot.py`

Engine

Simulation — `engine/simulation.py`

Agent — `engine/agent.py`

AgentRuntime — `engine/agent_runtime.py`

ActionRegistry — `engine/registry.py`

Schedulers — `engine/scheduler.py`

EventBus — `engine/event_bus.py`

Service

Session — `service/session.py`

DTOs — `service/dto.py`

Interfaces — `service/interfaces.py`

EventStreamAdapter — `service/events.py`

Transport

WebSocketServer — `transport/websocket/server.py`

Protocol constants — `transport/websocket/protocol.py`

Plugins

OllamaLLMBrain — `plugins/builtin/ollama/brain.py`

IdleBrain — `plugins/builtin/idle_brain/brain.py`

SimpleMemory — `plugins/builtin/simple_memory/memory.py`

ReplayBrain / JSONReplayRecorderSubscriber

HostedWorld — `plugins/external/world.py`

Config

CLI

Unity SDK (C#)

Class	Kind	Key members
`Scheduler`	Protocol	`async run_tick(agents, step_fn) -> list[tuple[Agent, Any]]`
`SimultaneousScheduler`	class	Runs all agents concurrently via `asyncio.gather`. Default scheduler.
`SequentialScheduler`	class	`__init__(order: list[str] \| None = None)` Runs agents one by one in declared order. Deterministic.

Class / constant	Kind	Key members
`Event`	dataclass	`type: str` `tick: int` `agent_id: str \| None` `data: dict`
`EventBus`	class	`subscribe(event_type, handler) -> None` `unsubscribe(event_type, handler) -> None` `emit(event) -> None`
Event type constants	str constants	`TICK_START`, `TICK_END` `ACTION_COMPLETED`, `ACTION_FAILED` `BRAIN_DECIDED` `AGENT_STEP_ERROR` `AGENT_REGISTERED`, `AGENT_UNREGISTERED`
`SocialEffectSubscriber`	class	Applies social `side_effects` from `ACTION_COMPLETED` events to `SocialSystem`.
`EventLogSubscriber`	class	Appends a formatted line to a bounded `deque(maxlen=1000)` for each `ACTION_COMPLETED` event. Access via `log_sub.tail(n)`.
`ObservabilitySubscriber`	class	Calls user-provided hooks: `on_tick`, `on_action`, `on_failure`, `on_event`. All hooks are optional.

Class / function	Kind	Key members
`SessionError`	exception	Raised when a method is called in an incompatible `SessionState` or `TickMode`.
`SimulationSession`	class	`session_id: str` (property) `state: SessionState` (property) `tick: int` (property) `tick_mode: TickMode` (property) `async step(req: StepRequest \| None) -> StepResponse` `async run() -> None` `pause() -> None` `resume() -> None` `shutdown() -> None` `subscribe(event_type, handler) -> str` (returns subscription_id) `unsubscribe(subscription_id: str) -> None` `register_agent(definition) -> Agent` `unregister_agent(agent_id) -> Agent` `snapshot() -> SimulationSnapshot` `restore(snap) -> None` `status() -> SimulationStatus`
`create_session(sim, tick_mode?) -> SimulationSession`	function	Factory. `tick_mode` defaults to `TickMode.HOST_DRIVEN`. Wires `EventStreamAdapter` to the simulation's `EventBus`.

Class	Kind	Key members
`AgentObservationDTO`	dataclass	`agent_id: str` `observation: dict`
`StepRequest`	dataclass	`agent_observations: list[AgentObservationDTO] = []` `world_metadata: dict =`
`AgentDecisionDTO`	dataclass	`agent_id: str` `agent_name: str` `action: str` `parameters: dict =` `outcome_text: str = ""` `engine_commands: list[dict] = []` `error: str \| None = None`
`StepResponse`	dataclass	`tick: int` `decisions: list[AgentDecisionDTO]` `errors: list[tuple[str, str]]` (agent_id, message) `engine_commands() -> list[dict]`
`ServiceEvent`	dataclass	`session_id: str` `event_type: str` `tick: int` `agent_id: str` `data: dict =`
`SimulationStatus`	dataclass	`session_id: str` `state: str` `tick: int` `config_ticks: int` `agent_count: int` `has_world_snap: bool` `tick_mode: str = "host_driven"`

Class	Kind	Members
`TickMode`	str enum	`HOST_DRIVEN = "host_driven"` \| `AUTONOMOUS = "autonomous"`
`SessionState`	str enum	`CREATED` \| `RUNNING` \| `PAUSED` \| `STOPPED` \| `ERROR`
`EventHandler`	type alias	`Callable[[ServiceEvent], None]`
`SimulationController`	Protocol	Structural type satisfied by `SimulationSession`. Declares all lifecycle + step methods.

Class	Members
`Method`	`HEALTH_CHECK`, `REGISTER_AGENT`, `REMOVE_AGENT`, `SEND_OBSERVATION`, `TICK`, `PAUSE`, `RESUME`, `SNAPSHOT`, `RESTORE`, `SUBSCRIBE_EVENTS`, `UNSUBSCRIBE_EVENTS`
`ErrorCode`	`PARSE_ERROR = -32700` `INVALID_REQUEST = -32600` `METHOD_NOT_FOUND = -32601` `INVALID_PARAMS = -32602` `INTERNAL_ERROR = -32603` `VERSION_MISMATCH = -32000` `SESSION_ERROR = -1` `AGENT_EXISTS = -2` `AGENT_NOT_FOUND = -3` `IMPORT_ERROR = -4` `SNAPSHOT_INVALID = -5`
`ProtocolError`	Exception with `code: int`, `name: str`, `message: str`

Function	Returns
`build_hello(session_id, tick_mode?) -> dict`	`hlo` frame — capabilities computed internally from tick_mode
`build_response(req_id, result) -> dict`	success `res` frame
`build_error(req_id, error) -> dict`	error `res` frame
`build_event(session_id, seq, service_event) -> dict`	`evt` frame

Class	Kind	Key members
`Personality`	dataclass	`traits: list[str]` `goals: list[str]` `backstory: str`
`OllamaLLMBrain`	class (Brain)	`__init__(personality?, llm_config?, system_prompt?, **kwargs)` `llm_config` keys: `model` (default `"qwen2.5:14b"`), `base_url` (default `"http://localhost:11434"`), `temperature`, `max_concurrent` `async decide(agent, observation, actions, context) -> Intent`

Class	Module	Key members
`ReplayBrain`	`plugins/builtin/replay_brain/brain.py`	`__init__(mode="replay", path="replay.jsonl", inner?, llm_config?, **kwargs)` `mode="record"`: wraps an inner brain and records all intents to JSONL. `mode="replay"`: reads intents from a JSONL file deterministically. Implements `Closeable` and `Snapshotable`.
`JSONReplayRecorderSubscriber`	`plugins/builtin/replay_recorder/recorder.py`	`__init__(path: str \| Path)` EventBus subscriber that writes one JSON record per event to a JSONL file. Subscribe to `BRAIN_DECIDED`, `ACTION_COMPLETED`, or `"*"`. Implements context manager; call `close()` when done.

Field	Type	Description
`engine.ticks`	int (default 20)	Total ticks to run (autonomous mode)
`engine.seed`	int (default 42)	RNG seed for determinism
`engine.scheduler`	`sequential` \| `simultaneous`	Agent execution order
`engine.scheduler_order`	list[str]	Optional explicit agent_id order for sequential scheduler
`engine.log_level`	`normal` \| `verbose` \| `quiet`	Logging verbosity
`world.class`	dotted path (required)	World implementation to instantiate
`registry.class`	dotted path	Callable returning a populated `ActionRegistry`. Optional — defaults to empty registry.
`observations.class`	dotted path	Optional callable returning an `ObservationRegistry`
`social.class`	dotted path	Optional SocialSystem implementation
`llm`	dict	Top-level LLM config dict forwarded to brain constructors (e.g. `model`, `base_url`)
`roles`	dict	Named role definitions — each with `capabilities` and `brain` block. Agents may reference a role via `agents[].role`.
`agents[].id`	str	Unique agent identifier
`agents[].name`	str	Display name
`agents[].role`	str	Reference to a declared role; inherits capabilities and brain
`agents[].position`	dict	Passed to `PlaceableWorld.place_agent()` if world supports it
`agents[].brain.class`	dotted path	Brain implementation (required unless role supplies one)
`agents[].state_ext.class`	dotted path	AgentStateExtension implementation
`agents[].capabilities`	list[str]	Capability tags (additive union with role capabilities)
`agents[].memory.class`	dotted path	Memory implementation (default: SimpleMemory)
`agents[].inventory`	dict	Starting item counts
`agents[].metadata`	dict	Arbitrary key/value data attached to agent; not consumed by tick loop

Class	Namespace / file	Role
`BiomataManager`	`Runtime/Unity/`	MonoBehaviour singleton. Inspector-configurable connection settings. Exposes `Client` property and C# events.
`SimulationClient`	`Runtime/Core/`	Main entry point. Properties: `Agents`, `Ticks`, `Observations`, `Events`, `Snapshots`. `ConnectAsync` / `DisconnectAsync`.
`BiomataConfig`	`Runtime/Core/`	Connection config: `Host`, `Port`, `UseTls`, `ConnectTimeoutSeconds`, `DefaultCallTimeoutSeconds`, `Retry`.
`AgentClient`	`Runtime/Clients/`	`RegisterAsync(AgentRegistration)`, `RemoveAsync(agentId)`, `TryRemoveAsync(agentId)`.
`TickClient`	`Runtime/Clients/`	`TickAsync(observations, worldMetadata, ct)`, `PauseAsync(ct)`, `ResumeAsync(ct)`.
`EventStreamClient`	`Runtime/Clients/`	Subscribe / unsubscribe to server-side event streams. Raises per-type C# events.
`SnapshotClient`	`Runtime/Clients/`	`SnapshotAsync(ct)`, `RestoreAsync(snapshot, ct)`.
`AgentRegistration`	`Runtime/Models/`	DTO: `AgentId`, `AgentName`, `BrainClass`, `BrainConfig`, `MemoryClass`, `MemoryConfig`.
`AgentObservationData`	`Runtime/Models/`	DTO: `AgentId`, `Observation` (Dictionary).
`TickResult`	`Runtime/Models/`	`Tick`, `Decisions` (`IReadOnlyList`), `Errors`, `AllCommands`, `ForAgent(agentId)` O(1).
`AgentDecisionResult`	`Runtime/Models/`	Per-agent decision: `AgentId`, `Action`, `Target`, `Reasoning`, `Success`, `Outcome`, `EngineCommands`.
`RetryConfig`	`Runtime/Core/`	`MaxAttempts`, `InitialDelaySeconds`, `MaxDelaySeconds`, `Multiplier`, `JitterFraction`.

API Reference

Contracts

Actions — contracts/action.py

World — contracts/world.py

Brain — contracts/brain.py

Observation — contracts/observation.py

ObservationRegistry — engine/obs_registry.py

Memory — contracts/memory.py

State — contracts/state.py

Social — contracts/social.py

Snapshot — contracts/snapshot.py

Engine

Simulation — engine/simulation.py

Agent — engine/agent.py

AgentRuntime — engine/agent_runtime.py

ActionRegistry — engine/registry.py

Schedulers — engine/scheduler.py

EventBus — engine/event_bus.py

Service

Session — service/session.py

DTOs — service/dto.py

Interfaces — service/interfaces.py

EventStreamAdapter — service/events.py

Transport

WebSocketServer — transport/websocket/server.py

Protocol constants — transport/websocket/protocol.py

Plugins

OllamaLLMBrain — plugins/builtin/ollama/brain.py

IdleBrain — plugins/builtin/idle_brain/brain.py

SimpleMemory — plugins/builtin/simple_memory/memory.py

ReplayBrain / JSONReplayRecorderSubscriber

WeightedGraphSocial — plugins/builtin/simple_social/social.py

HostedWorld — plugins/external/world.py

Config

CLI

Unity SDK (C#)

Actions — `contracts/action.py`

World — `contracts/world.py`

Brain — `contracts/brain.py`

Observation — `contracts/observation.py`

ObservationRegistry — `engine/obs_registry.py`

Memory — `contracts/memory.py`

State — `contracts/state.py`

Social — `contracts/social.py`

Snapshot — `contracts/snapshot.py`

Simulation — `engine/simulation.py`

Agent — `engine/agent.py`

AgentRuntime — `engine/agent_runtime.py`

ActionRegistry — `engine/registry.py`

Schedulers — `engine/scheduler.py`

EventBus — `engine/event_bus.py`

Session — `service/session.py`

DTOs — `service/dto.py`

Interfaces — `service/interfaces.py`

EventStreamAdapter — `service/events.py`

WebSocketServer — `transport/websocket/server.py`

Protocol constants — `transport/websocket/protocol.py`

OllamaLLMBrain — `plugins/builtin/ollama/brain.py`

IdleBrain — `plugins/builtin/idle_brain/brain.py`

SimpleMemory — `plugins/builtin/simple_memory/memory.py`

WeightedGraphSocial — `plugins/builtin/simple_social/social.py`

HostedWorld — `plugins/external/world.py`