Python SDK

The Python SDK (sdk/py/) provides pytest integration and scripting support for interacting with world-runner from Python code.

Requirements

Python ≥ 3.12
plant_world_runner binary built with server + brand-plugin features

Setup

cd sdk/py
python3 -m venv .venv
source .venv/bin/activate
pip install -e ".[test]"

Running tests

pytest

The binary path defaults to ../../target/debug/plant_world_runner. Override with WORLD_RUNNER_BIN.

WorldRunnerHarness

src/plantangenet/harness.py — spawn and terminate world-runner for testing.

from plantangenet import WorldRunnerHarness

# Explicit lifecycle
harness = WorldRunnerHarness()
harness.start()
client = harness.make_client()
harness.stop()

# Context manager
with WorldRunnerHarness() as harness:
    client = harness.make_client()

Constructor parameters

Parameter	Type	Default	Description
`binary`	`str \\| None`	`../../target/debug/plant_world_runner`	Path to binary (or `WORLD_RUNNER_BIN` env)
`admin_email`	`str`	`test@harness.local`	Email for admin magic-link session
`extra_args`	`list[str]`	`[]`	Additional CLI args for world-runner

Properties

Property	Type	Description
`base_url`	`str`	e.g. `http://127.0.0.1:43829`
`port`	`int`	Bound TCP port
`session_cookie`	`str`	Raw `key=value` session cookie

Methods

Method	Returns	Description
`start()`	`self`	Spawn, wait for ready, bootstrap auth
`make_client()`	`WorldRunnerClient`	Authenticated client for this session
`stop()`	`None`	SIGTERM + drain

WorldRunnerClient

src/plantangenet/world_client.py — authenticated HTTP + WebSocket client.

from plantangenet import WorldRunnerClient

client = WorldRunnerClient('http://127.0.0.1:8080', session_cookie)

Clock

Method	Description
`status()`	`GET /status` — clock/frame status (no auth required)
`clock_step()`	`POST /clock {"action":"step"}` — advance one tick
`clock_set_interval(ms)`	Start continuous clock at `ms` milliseconds/tick
`clock_pause()`	Pause continuous clock
`clock_resume()`	Resume paused clock

Atoms

Method	Description
`get_atom_kinds()`	`GET /atoms` — list of kind strings
`get_atoms(kind)`	`GET /atoms/:kind` — all instances of a kind
`get_atom(kind, id)`	`GET /atoms/:kind/:id` — single atom, raises on 404

Services

Method	Description
`get_services()`	`GET /services` — service descriptor list
`get_service(kind)`	`GET /services/:kind`
`get_service_registry(kind)`	`GET /services/:kind/registry`

WebSocket stream

import json

# Open a stream (optionally filtered)
ws = client.open_stream(prefix='hob/', entity_id='hob-1', timeout=10.0)

# Block until a matching event arrives
event = client.wait_for_stream_event(
    ws,
    lambda e: e.get('event') == 'heartbeat',
    timeout=5.0,
)

ws.close()

websocket-client is used so the session cookie can be forwarded on the HTTP upgrade request, satisfying the server's RequireAuth guard.

HobDriver

src/plantangenet/hob_driver.py — high-level per-hob wrapper.

from plantangenet import HobDriver

driver = HobDriver(client, 'hob-1')

Reading state

state = driver.get_state()
# → raw atom dict from /atoms/hob/hob-1

ids = HobDriver.list_hobs(client)
# → ['hob-1', 'hob-2', ...]

Observing the stream

# Open an entity-filtered WebSocket
ws = driver.open_stream(timeout=10.0)

# Step clock + wait for any event (returns event dict or None on timeout)
event = driver.step_and_observe(timeout=3.0)

Phase II movement (stubs)

The following methods are planned once world-runner exposes hob input patch endpoints:

walk_to_node(node_id)
walk_along_segment(segment_id)
set_stance(stance) — 'idle' | 'walking' | 'stationary_watch'
await_location(predicate, timeout=30.0)

Pytest fixtures

tests/conftest.py provides module-scoped fixtures so a single world-runner process serves all tests in a module:

# tests/conftest.py (already provided)
import pytest
from plantangenet import WorldRunnerHarness, WorldRunnerClient

@pytest.fixture(scope="module")
def harness():
    h = WorldRunnerHarness()
    h.start()
    yield h
    h.stop()

@pytest.fixture(scope="module")
def client(harness) -> WorldRunnerClient:
    return harness.make_client()

Use them in any test file:

def test_clock_step(client):
    result = client.clock_step()
    assert isinstance(result, dict)