Protocol Contracts

Contracts

Contracts are the stable extension points of Biomata Engine. They live in src/contracts/ and are defined as Python Protocol classes with @runtime_checkable. Implementations do not need to inherit from anything — they just need to satisfy the structural type.

Protocol typing

All contracts use typing.Protocol with @runtime_checkable. This means isinstance(obj, World) works at runtime and mypy checks structural compatibility statically. You never inherit from these — just implement the required methods.

Actions

The action system defines how agents express decisions (Intent), how those decisions are executed (ActionHandler), what results look like (ActionResult), and how actions are described to the brain (ActionSchema). Source: src/contracts/action.py.

Intent

Intent is the output of Brain.decide(). It expresses what an agent wants to do.

@dataclass class Intent: action: str = "idle" target: str | None = None # agent_id of the target, or None parameters: dict[str, Any] = # action-specific params reasoning: str = "" # one-sentence explanation (from LLM)

StateMutations

The typed container for Python-side mutation channels. Replaces the older state_mutations dict.

@dataclass class StateMutations: inventory: dict[str, int] = # item count deltas; clamped at 0 ext: dict[str, Any] = # forwarded to state_ext.apply_mutations()

ActionResult

ActionResult is returned by ActionHandler.execute(). It carries state changes as data rather than imperative mutations. Source: src/contracts/action.py.

@dataclass class ActionResult: success: bool outcome_text: str # human-readable log line; stored in memory mutations: StateMutations = StateMutations() side_effects: list[dict] = [] # social effects, future extensibility engine_commands: list[dict] = [] # relayed to host engine (Unity/Unreal)

Legacy state_mutations

Passing state_mutations={...} still works — the dict is automatically converted to a StateMutations object in __post_init__. The inventory key maps to mutations.inventory; all other keys map to mutations.ext. New code should use mutations=StateMutations(...) directly.

mutations conventions

Channel	Meaning
`mutations.ext = {"energy_delta": -5}`	Forwarded verbatim to `state_ext.apply_mutations()` — semantics are state_ext-defined
`mutations.inventory = {"food": 3}`	Integer deltas applied to `agent.inventory`; each item clamped to 0

engine_commands conventions

Commands are opaque dicts collected by HostedWorld.apply() and forwarded to the host. The "type" key is required; all other keys are handler-defined.

python

from src.contracts.action import ActionResult, StateMutations

return ActionResult(
    success=True,
    outcome_text="moving north",
    mutations=StateMutations(ext={"energy_delta": -5}),
    engine_commands=[
        {"type": "navigate", "direction": "north"},
        {"type": "set_animation", "clip": "walk"},
    ],
)

side_effects for social updates

python

return ActionResult(
    success=True,
    outcome_text="shared food with Mira",
    side_effects=[
        {"type": "social", "from": "agent_001", "to": "agent_002", "delta": 0.3}
    ],
)

ActionHandler

class ActionHandler(Protocol): def execute( self, agent: AgentView, intent: Intent, context: WorldContext, ) -> ActionResult: ...

Implement one class per action. Handlers are pure: they do not mutate world state — they return mutations as data. The engine applies them.

python

class GossipHandler:
    def execute(self, agent, intent, context) -> ActionResult:
        target_id = intent.target
        if target_id is None:
            return ActionResult(success=False, outcome_text="no one to gossip with")

        target = context.get_agent(target_id)
        if target is None:
            return ActionResult(success=False, outcome_text=f"{target_id} not found")

        from src.contracts.world import SpatialWorld
        if isinstance(context, SpatialWorld) and not context.are_adjacent(agent.id, target_id):
            return ActionResult(success=False, outcome_text="too far to gossip")

        return ActionResult(
            success=True,
            outcome_text=f"gossiped with {target.name}",
            side_effects=[
                {"type": "social", "from": agent.id, "to": target_id, "delta": 0.2}
            ],
        )

ActionHint (execution_hint)

Advisory label that tells the LLM who is responsible for executing the action. Does not change dispatch logic. Source: src/contracts/action.py.

class ActionHint(str, Enum): HOST = "host" # Unity/renderer executes; Python packages engine_commands only ENGINE = "engine" # Python executes; may mutate world state, inventory, social graph HYBRID = "hybrid" # both channels; default — no label appended to prompt ActionKind = ActionHint # backwards-compatible alias

The hint label is appended to the action's prompt block so the LLM understands the execution boundary. HYBRID is the default.

ActionSchema

@dataclass class ActionSchema: name: str description: str parameters_schema: dict[str, Any] = # shown to LLM in prompt execution_hint: ActionHint = ActionHint.HYBRID required_capabilities: frozenset[str] = frozenset() # capability filter (see below) example: dict | None = None # single example dict for prompt

required_capabilities controls per-agent capability filtering — only agents whose capabilities intersect a schema's required_capabilities will see it. Schemas with no required capabilities are universal (visible to all agents).

example supplies a concrete JSON sample that prompt_block() injects immediately after the parameter list.

Deprecated aliases

kind= is a deprecated alias for execution_hint=. tags= is a deprecated alias for required_capabilities=. examples=[...] (list form) is a deprecated alias for example={...} (single dict). All three aliases still work.

python

from src.contracts.action import ActionHint, ActionSchema

# HOST action — Unity/renderer moves the character
registry.register(
    ActionSchema(
        "navigate",
        "Move to a world-space XZ position.",
        {"target_x": "float", "target_z": "float"},
        execution_hint=ActionHint.HOST,
        example={"action": "navigate", "parameters": {"target_x": 14.0, "target_z": 0.0}},
    ),
    NavigateHandler(),
)

# ENGINE action — Python computes the outcome directly
registry.register(
    ActionSchema("idle", "Stand still and wait.", execution_hint=ActionHint.ENGINE),
    IdleHandler(),
)

# Capability-gated action — only agents with matching capabilities will see this
registry.register(
    ActionSchema(
        "cast_spell",
        "Cast a magical spell at your current location.",
        {"spell": "fireball|heal|shield"},
        execution_hint=ActionHint.HOST,
        required_capabilities=frozenset({"mage"}),
        example={"action": "cast_spell", "parameters": {"spell": "fireball"}},
    ),
    CastSpellHandler(),
)

Per-agent capability filtering

By default every agent sees every registered action. Use required_capabilities on the schema and capabilities on the agent to restrict which actions are offered to specific agents.

Visibility rule: a schema is visible to an agent if the schema's required_capabilities is empty (universal) or if the agent's capabilities intersect required_capabilities.

Schema required_capabilities	Agent capabilities	Visible?
`frozenset()`	anything	yes — universal
`{"guard"}`	`{"guard"}`	yes
`{"guard"}`	`frozenset()`	no
`{"guard", "military"}`	`{"guard"}`	yes — intersection

Set capabilities on an agent in Python or YAML:

python

agent = Agent(
    id="guard_001",
    name="Aldric",
    brain=...,
    memory=...,
    capabilities=frozenset({"guard", "military"}),
)

yaml

agents:
  - id: mage_001
    name: Seraphel
    capabilities: [mage]
    brain:
      class: src.plugins.builtin.ollama.brain.OllamaLLMBrain
      ...

AgentRuntime calls registry.schemas_for(agent.capabilities) on each step, so the brain's action list is already filtered — it only ever sees actions it is permitted to choose from.

LLM prompt integration

ActionSchema.prompt_block() renders one action entry for the system prompt. The actions list passed to Brain.decide() is already filtered to this agent's capabilities by AgentRuntime — call prompt_block() on each schema to build the system prompt section.

Example output of a rendered action list:

text

AVAILABLE ACTIONS (use exactly these names):
  navigate: Move to a world-space XZ position.  [host]
    params: {"target_x":"float","target_z":"float"}
    example: {"action":"navigate","parameters":{"target_x":0.0,"target_z":0.0}}
  idle: Stand still and wait.  [engine]
  speak: Say something aloud (short phrase, under 15 words).  [host]
    params: {"message":"string — what you say"}
    example: {"action":"speak","parameters":{"message":"Fresh goods today!"}}

OllamaLLMBrain does this automatically. Any custom LLM brain should do the same:

python

async def decide(self, agent, observation, actions, context) -> Intent:
    # `actions` is already filtered to this agent's capabilities by AgentRuntime
    action_lines = ["AVAILABLE ACTIONS (use exactly these names):"]
    for schema in actions:
        action_lines.append(schema.prompt_block())
    system_prompt = f"You are {agent.name}.\n\n" + "\n".join(action_lines)
    ...

ActionValidationError

Structured error produced by ActionSchema.validate_parameters() and ActionRegistry.validate_intent(). Source: src/contracts/action.py.

@dataclass class ActionValidationError: code: str # "unknown_action" | "capability_denied" | "missing_param" | "type_mismatch" message: str field: str | None = None # parameter name, if applicable

HOST actions and engine_commands

A HOST action's handler should populate engine_commands in the returned ActionResult. The WebSocket transport forwards these to Unity (or whatever host is connected) as part of the tick response. Commands are opaque — any JSON-serialisable dict is valid; define the shape in your handler and mirror it in your Unity ActionHandlerBase.

python

class NavigateHandler:
    def execute(self, agent, intent, context) -> ActionResult:
        p = intent.parameters or {}
        tx, tz = float(p.get("target_x", 0)), float(p.get("target_z", 0))
        return ActionResult(
            success=True,
            outcome_text=f"navigate to ({tx:.1f}, {tz:.1f})",
            engine_commands=[{"type": "navigate", "x": tx, "y": 0.0, "z": tz}],
        )

Intent parsing

parse_intent(raw, valid_actions) converts raw LLM JSON output into an Intent. It handles JSON wrapped in markdown fences, partial JSON, and falls back to keyword matching.

World

The World protocol is the minimal interface the engine requires. Additional capability protocols allow handlers and the engine to declare precisely what they need. Source: src/contracts/world.py.

World protocol

class World(Protocol): def observe(self, agent_id: str) -> dict[str, Any]: ... def apply(self, agent_id: str, result: ActionResult) -> None: ... def tick(self) -> None: ... @property def current_tick(self) -> int: ... @property def metadata(self) -> dict[str, Any]: ...

WorldContext protocol

Passed to ActionHandler.execute() as the context argument. Handlers use this for reads — they never mutate the world.

class WorldContext(Protocol): rng: random.Random # canonical seeded RNG injected by Simulation def get_agent(self, agent_id: str) -> AgentView | None: ... def get_world_data(self) -> dict[str, Any]: ...

AgentView

@dataclass class AgentView: id: str name: str inventory: dict[str, Any] ext: dict[str, Any] = # state_ext snapshot

Optional capability protocols

Protocol	Methods	Used by
`SpatialWorld`	`are_adjacent(id1, id2) -> bool`	Handlers that need proximity checks (speak, trade, attack). Also satisfies `WorldContext`.
`VisibilityWorld`	`get_nearby_agents(agent_id) -> list[AgentView]`	Engine builds `nearby_agents` field in observations. Returns `AgentView` objects, not strings.
`PlaceableWorld`	`place_agent(agent_id, **kwargs)`	Config loader — applies YAML `position:` fields at startup
`ExternalWorld`	`push_observation()`, `push_metadata()`, `collect_commands()`	Host-authoritative mode — game engine owns world state. Extends `World`.

ExternalWorld

Extends the World protocol for host-authoritative worlds. The game engine pushes observations in; Python runs cognition; commands flow back out. Check with isinstance(world, ExternalWorld).

python

from src.contracts.world import ExternalWorld
from src.plugins.external.world import HostedWorld

world = HostedWorld()
assert isinstance(world, ExternalWorld)   # True

# Per-tick cycle
world.push_metadata({"time_of_day": "noon"})
world.push_observation("agent_001", {"location": "gatehouse", "health": 100})
summary = await sim.run_tick()
commands = world.collect_commands()   # [{"agent_id": "agent_001", "type": "navigate", ...}]

Implementing a custom world

python

import random
from src.contracts.action import ActionResult
from src.contracts.world import AgentView

class IslandWorld:
    """Simple island world with named locations."""

    def __init__(self):
        self.rng: random.Random = random.Random()
        self._tick = 0
        self._positions: dict[str, str] = {}

    def observe(self, agent_id: str) -> dict[str, Any]:
        return {"location": self._positions.get(agent_id, "beach")}

    def apply(self, agent_id: str, result: ActionResult) -> None:
        if "move_to" in result.state_mutations:
            self._positions[agent_id] = result.state_mutations["move_to"]

    def tick(self) -> None:
        self._tick += 1

    @property
    def current_tick(self) -> int:
        return self._tick

    @property
    def metadata(self) -> dict[str, Any]:
        return {"tick": self._tick}

    def get_agent(self, agent_id: str) -> AgentView | None:
        return None

    def get_world_data(self) -> dict[str, Any]:
        return {"locations": ["beach", "jungle", "cave"]}

Brain

The Brain protocol defines cognition. A Brain receives the agent's current state and context, and returns an Intent. Personality, backstory, prompt templates, and LLM calls all live in the Brain — not in Agent. Source: src/contracts/brain.py.

class Brain(Protocol): async def decide( self, agent: AgentView, observation: dict[str, Any], actions: list[ActionSchema], context: BrainContext, ) -> Intent: ...

BrainContext

@dataclass class BrainContext: tick: int memory: str = "" # formatted recall from Memory.recall() metadata: dict[str, Any] = # world.metadata — tick, season, weather, etc. observation_schemas: list[ObservationSchema] = [] # schemas visible to this agent emit: Any = None # EventBus.emit — optional, for BRAIN_DECIDED events

Social relationship data is no longer a separate BrainContext field. Instead it is injected directly into the agent's observation dict via a SocialContextProvider registered in the ObservationRegistry. The brain sees it as observation["social_relationships"].

Custom brain example

python

import random
from src.contracts.action import Intent, ActionSchema
from src.contracts.brain import BrainContext
from src.contracts.world import AgentView

class UtilityBrain:
    """Chooses the action with the highest utility score."""

    def __init__(self, utilities: dict[str, float] | None = None):
        self._utilities = utilities or {}

    async def decide(
        self,
        agent: AgentView,
        observation: dict,
        actions: list[ActionSchema],
        context: BrainContext,
    ) -> Intent:
        best_action = "idle"
        best_score  = -float("inf")

        for schema in actions:
            score = self._utilities.get(schema.name, 0.0)
            if schema.name == "gather_food" and observation.get("food_nearby"):
                score += 10.0
            if score > best_score:
                best_score  = score
                best_action = schema.name

        return Intent(action=best_action, reasoning=f"utility={best_score:.1f}")

Closeable

Optional lifecycle protocol for brains that hold external resources (open file handles, network connections, worker threads). The engine calls close() on teardown for every agent whose brain implements this protocol. Source: src/contracts/brain.py.

@runtime_checkable class Closeable(Protocol): def close(self) -> None: ... # must be idempotent

Implement it when your brain allocates resources that need explicit cleanup:

python

class StreamingLLMBrain:
    def __init__(self, ...):
        self._session = aiohttp.ClientSession()

    async def decide(self, agent, observation, actions, context) -> Intent:
        ...

    def close(self) -> None:
        asyncio.get_event_loop().run_until_complete(self._session.close())

Use the Simulation as a context manager to guarantee close() is called even if the simulation raises:

python

with Simulation(agents=[...], ...) as sim:
    await sim.run()
# All Closeable brains have been closed here

Or call sim.close() explicitly. Non-Closeable brains are silently skipped; a brain that raises from close() logs a warning and the remaining brains are still closed.

Memory

Per-agent episodic memory. The engine calls store() after each step and recall() to build brain context. Source: src/contracts/memory.py.

class Memory(Protocol): def store(self, tick: int, observation: str, intent: Intent, outcome: str) -> None: ... def recall(self, n: int = 6) -> str: ... # n most recent as formatted string def serialize(self) -> bytes: ... def restore(self, data: bytes) -> None: ...

The default implementation is SimpleMemory — a rolling deque of formatted strings. Custom implementations can use vector search, symbolic graphs, or any other storage.

AgentStateExtension

Domain-specific per-agent state (vitals, stress, energy, pheromone levels). The engine calls tick() automatically each step and apply_mutations() with the handler's returned mutations. Source: src/contracts/state.py.

class AgentStateExtension(Protocol): def tick(self) -> None: ... # decay, regen, etc. def apply_mutations(self, mutations: dict[str, Any]) -> None: ... def snapshot(self) -> dict[str, Any]: ... # read-only dict for observations def to_prompt_str(self) -> str: ... # formatted string for brain prompt def urgent_advice(self) -> str: ... # one-line warning for LLM, or "" def serialize(self) -> bytes: ... def restore(self, data: bytes) -> None: ...

Custom state extension example

python

import pickle

class SpaceshipVitals:
    def __init__(self, hull: int = 100, fuel: int = 100):
        self.hull = hull
        self.fuel = fuel

    def tick(self) -> None:
        self.fuel = max(0, self.fuel - 1)

    def apply_mutations(self, mutations: dict) -> None:
        if "hull_delta" in mutations:
            self.hull = max(0, min(100, self.hull + mutations["hull_delta"]))
        if "fuel_delta" in mutations:
            self.fuel = max(0, min(100, self.fuel + mutations["fuel_delta"]))

    def snapshot(self) -> dict:
        return {"hull": self.hull, "fuel": self.fuel}

    def to_prompt_str(self) -> str:
        return f"Hull: {self.hull}/100 | Fuel: {self.fuel}/100"

    def urgent_advice(self) -> str:
        if self.fuel < 10:
            return "CRITICAL: fuel nearly depleted — find a fuel depot immediately!"
        if self.hull < 20:
            return "WARNING: hull integrity critical — avoid combat!"
        return ""

    def serialize(self) -> bytes:
        return pickle.dumps({"hull": self.hull, "fuel": self.fuel})

    def restore(self, data: bytes) -> None:
        state = pickle.loads(data)
        self.hull = state["hull"]
        self.fuel = state["fuel"]

SocialSystem

Tracks inter-agent relationships. Updated automatically via side_effects in ActionResult — handlers never call the social system directly. Source: src/contracts/social.py.

class SocialSystem(Protocol): def add_agent(self, agent_id: str, name: str) -> None: ... def update(self, from_id: str, to_id: str, delta: float) -> None: ... def relationship(self, from_id: str, to_id: str) -> float: ... def describe(self, agent_id: str) -> str: ... # human-readable summary for brain def serialize(self) -> bytes: ... def restore(self, data: bytes) -> None: ...

The built-in implementation is WeightedGraphSocial — a directed graph with float weights in the range [-1.0, +1.0]. It is configured in YAML as src.plugins.builtin.simple_social.social.WeightedGraphSocial.

Snapshot protocol

Components that implement Snapshotable are included in simulation snapshots. Source: src/contracts/snapshot.py.

class Snapshotable(Protocol): def serialize(self) -> bytes: ... # capture all mutable runtime state def restore(self, data: bytes) -> None: ...

Rules: serialize only mutable runtime state (not immutable config). Never include references to other simulation objects — Simulation.restore() re-wires those after component restoration. restore() must be idempotent.

Observations

The observation system lets you register named data slices that are merged into every agent's observation dict before Brain.decide() is called. It mirrors the action contract architecture — define a schema, implement a provider, register both. Source: src/contracts/observation.py, registry: src/engine/obs_registry.py.

ObservationSchema

Documents one named observation slot. The schema is rendered into the LLM system prompt so the brain understands what each field means.

@dataclass class ObservationSchema: name: str description: str payload_schema: dict[str, Any] = # field-name → type hint, shown in prompt required_capabilities: frozenset[str] = frozenset() # empty = universal; non-empty = capability-gated examples: list[dict] = [] # examples[0] shown in prompt

Deprecated alias

tags= is a deprecated alias for required_capabilities= and still works.

ObservationProvider

Produces a dict slice for one agent each tick. The returned dict is merged into the agent's observation. Source: src/contracts/observation.py.

The ObservationProvider protocol defines a collect() method. The built-in providers (SimulationTimeProvider, SocialContextProvider, etc.) implement observe() — both names are used in the codebase. When writing custom providers, implement the method that the registry will call, which is collect() per the protocol definition.

class ObservationProvider(Protocol): def collect( self, agent_id: str, capabilities: frozenset[str], world: World, ) -> dict[str, Any]: ...

Providers must handle their own exceptions — the registry logs and skips any provider that raises, so one broken provider cannot crash a tick.

ObservationRegistry

Central registry mapping observation names → (schema, provider). Architecture mirrors ActionRegistry. Source: src/engine/obs_registry.py.

python

from src.contracts.observation import ObservationSchema
from src.engine.obs_registry import ObservationRegistry

obs_registry = ObservationRegistry()

obs_registry.register(
    ObservationSchema(
        "nearby_agents",
        "Agents within sensor range.",
        {"id": "str", "name": "str", "distance": "float"},
        required_capabilities=frozenset({"social"}),   # only agents with "social" capability see this
    ),
    NearbyAgentsProvider(radius=8.0),
)

# Merge all matching providers into the observation dict for agent_001
merged = obs_registry.collect("agent_001", agent.capabilities, world)

Method	Description
`register(schema, provider)`	Add a (schema, provider) pair. Raises `ValueError` on duplicate name.
`schemas()`	All registered `ObservationSchema` objects.
`schemas_for(capabilities)`	Schemas visible to an agent with the given capability set.
`collect(agent_id, capabilities, world)`	Run all matching providers, merge slices; exceptions are logged and skipped.
`observations_prompt_section(capabilities?)`	Render visible schemas as a prompt section for LLM system prompts.

Built-in providers live in src/plugins/builtin/observations/providers.py:

Provider	Produces
`SimulationTimeProvider`	`simulation_tick: int`
`SocialContextProvider(social)`	`social_relationships: str` — calls `social.describe(agent_id)`
`IncomingMessagesProvider(inbox)`	`incoming_messages: list[{"from", "text"}]` — consumed on delivery
`FunctionProvider(fn)`	Wraps any `lambda agent_id, world: dict` as a provider

Contracts

Actions

Intent

StateMutations

ActionResult

mutations conventions

engine_commands conventions

side_effects for social updates

ActionHandler

ActionHint (execution_hint)

ActionSchema

Per-agent capability filtering

LLM prompt integration

ActionValidationError

HOST actions and engine_commands

Intent parsing

World

World protocol

WorldContext protocol

AgentView

Optional capability protocols

ExternalWorld

Implementing a custom world

Brain

BrainContext

Custom brain example

Closeable

Memory

State & Social

AgentStateExtension

Custom state extension example

SocialSystem

Snapshot protocol

Observations

ObservationSchema

ObservationProvider

ObservationRegistry