Overview

Universtato is a persistent, networked 3D world model server — the shared spatial ground truth for AI agents, robots, and vision pipelines. Think of it as Redis for 3D state: agents read and write a live scene graph, lock regions before acting, simulate moves in isolated forks, and subscribe to spatial events without polling.

The server exposes two transports:

Connect

Python SDK (recommended)

from sdk.client import UniverstatoClient

async with UniverstatoClient(
    host="localhost",
    port=50051,      # gRPC
    http_port=8080, # HTTP / forks / webhooks
) as wm:
    # All operations here
    objects = await wm.list()

Raw HTTP

# Scene snapshot
GET http://localhost:8080/state

# Natural language command
POST http://localhost:8080/nl
Content-Type: application/json

{ "text": "put a red chair at the center of the room" }

Object Model

Every entity in the world is a SceneObject. This is the canonical JSON representation:

{
  "id":       "3f7a2c1d-...",   // UUID, immutable
  "name":     "oak_chair",
  "type":     "chair",          // semantic class; free-form string
  "version":  4,               // incremented on every mutation

  "transform": {
    "position": { "x": 1.5, "y": 0.0, "z": -2.0 },
    "rotation": { "x": 0.0, "y": 0.0, "z": 0.0, "w": 1.0 }, // quaternion
    "scale":    { "x": 1.0, "y": 1.0, "z": 1.0 }
  },

  "geometry": {
    "type":       "BOX",          // BOX | SPHERE | CYLINDER | CAPSULE | PLANE
    "dimensions": { "x": 0.6, "y": 0.9, "z": 0.6 },
    "radius":     0.0,
    "height":     0.0
  },

  "color": { "r": 0.6, "g": 0.3, "b": 0.1, "a": 1.0 },  // RGBA [0,1]

  "affordances": ["sit", "move", "stack"],  // what agents can do with this

  "physics": {
    "mass":            8.0,   // kg; 0 = kinematic
    "friction":        0.5,   // Coulomb [0, 1]
    "restitution":     0.1,   // bounciness [0, 1]
    "linear_damping":  0.05,
    "angular_damping": 0.05,
    "is_static":       false  // true = immovable (walls, floors)
  },

  "metadata": {            // arbitrary string key-value pairs
    "material": "wood",
    "agent_id": "planner-1"
  },

  "created_at": 1714512000.0,
  "updated_at": 1714512045.3
}

Place Objects

# Minimal
chair = await wm.place("chair", "oak_chair", position=(1.5, 0, -2.0))

# Full options
table = await wm.place(
    "table", "dining_table",
    position=(0, 0, 0),
    dimensions=(2.0, 0.75, 1.0),   # width × height × depth
    color=(0.8, 0.7, 0.5),
    metadata={"material": "oak", "seats": "4"},
)

HTTP equivalent

POST /forks/{fork_id}/place  # or use gRPC PlaceObject
{
  "type": "table",
  "name": "dining_table",
  "position": { "x": 0, "y": 0, "z": 0 },
  "geometry": { "type": "BOX", "dimensions": { "x": 2, "y": 0.75, "z": 1 } },
  "affordances": ["eat_at", "place_on"]
}

Query the Scene

List all objects

all_objects = await wm.list()

# Filter by type
chairs = await wm.list(type="chair")

# Filter by metadata
wooden = await wm.list(metadata_key="material", metadata_value="wood")

Spatial radius query

# Everything within 3m of the origin
nearby = await wm.query_spatial(center=(0, 0, 0), radius=3.0)

# Everything within 2m of a known object (by ID)
near_table = await wm.query_spatial(
    anchor_object_id=table["id"],
    radius=2.0,
)

Fetch one object

obj = await wm.get(object_id)
if obj:
    print(obj["name"], obj["transform"]["position"])

Full snapshot

snap = await wm.state()
# snap["objects"]  → list of all objects
# snap["sequence"] → monotonic counter, useful for change detection

Move & Update

# Move an object to a new position
updated = await wm.move(chair["id"], position=(2.0, 0, 1.0))

# Update metadata
updated = await wm.update(chair["id"], metadata={"occupied": "true"})

Remove

await wm.remove(chair["id"])
# The object is deleted from the scene graph and all subscribers are notified.

Affordances

Affordances are the list of actions an agent can take on an object. They are free-form strings that you define to match your agent's action vocabulary.

# Place an object with affordances
door = await wm.place(
    "door", "front_door", position=(5, 1, 0),
    metadata={"affordances": "open,close,lock,unlock"},
)

# Query by affordance — find everything an agent can sit on
sittable = await wm.list(metadata_key="affordances")
sittable = [o for o in sittable if "sit" in o.get("affordances", [])]

Common affordance vocabularies to consider:

Physics

The physics engine (optional, start with --physics) runs a 60Hz simulation loop. Physics properties on each object control simulation behaviour.

# Heavy, high-friction, static wall
wall = await wm.place(
    "wall", "north_wall", position=(0, 1.5, -5),
    dimensions=(10, 3, 0.2),
    metadata={"static": "true"},   # or set is_static via gRPC UpdateObject
)

# Lightweight bouncy ball
ball = await wm.place("sphere", "rubber_ball", position=(0, 3, 0))

Spatial Index

The scene graph maintains a 2-metre spatial hash grid for fast proximity lookups. All spatial queries are O(k) where k is the number of objects in the query radius, not total scene size.

# Radius query — returns objects sorted by distance
results = await wm.query_spatial(center=(3, 0, 3), radius=5.0)

# Anchor query — "what's within 2m of this robot?"
robot_id = "..."
nearby = await wm.query_spatial(anchor_object_id=robot_id, radius=2.0)

Each result object includes its full transform, affordances, physics, and metadata. For distance values, use the HTTP endpoint directly:

POST /forks/{id}/query
{ "anchor_object_id": "...", "radius": 2.0 }

→ { "objects": [...], "distances": [0.4, 1.1, 1.8] }

Spatial Locks

When multiple agents operate in the same world, spatial locks prevent conflicting writes. A lock is a 3D bounding box lease with a TTL (max 30s). The scene graph enforces locks: any place() or move() that would put an object inside a locked region raises a conflict.

# Acquire a lock before manipulating a region
lock_id = await wm.acquire_lock(
    agent_id="planner-agent-1",
    bbox_min=(-1, 0, -1),
    bbox_max=(1, 2, 1),
    ttl=10.0,   # seconds; max 30
)

try:
    await wm.place("table", "center_table", position=(0, 0, 0))
finally:
    await wm.release_lock(lock_id)  # always release early if done

# Handle conflict gracefully
from universtato.scene.locks import SpatialLockConflict

try:
    lock_id = await wm.acquire_lock("agent-2", bbox_min, bbox_max, ttl=5)
except RuntimeError as e:
    print(f"Region locked: {e}")  # retry or pick another region

Semantic Triggers

Triggers are persistent zone rules. When an object enters or exits a defined region, a TriggerFired event is dispatched to all subscribers. Useful for proximity alerts, zone access control, and reactive agent logic.

Register a zone

# Sphere zone — fire when anything enters within 1.5m of the door
rule_id = await wm.register_trigger(
    event_name="near_door",
    agent_id="door-agent",
    center=(5, 1, 0),
    radius=1.5,
    on_enter=True,
    on_exit=True,
)

# Bounding box zone — restricted area
rule_id = await wm.register_trigger(
    event_name="restricted_zone",
    bbox_min=(-3, 0, -3),
    bbox_max=(3, 3, 3),
    type_filter="robot",     # only fire for robots
    on_enter=True,
)

Subscribe to events (gRPC stream)

async def on_trigger(event):
    print(f"[{event['transition']}] {event['object_name']} → {event['event_name']}")
    print(f"  position: {event['position']}")

await wm.subscribe_triggers(
    callback=on_trigger,
    agent_id="door-agent",    # filter to your rules
    event_name="near_door",  # optional filter
)

The event dict delivered to the callback:

{
  "rule_id":     "...",
  "event_name":  "near_door",
  "transition":  "entered",  // or "exited"
  "object_id":   "...",
  "object_name": "robot_1",
  "object_type": "robot",
  "position":    { "x": 4.8, "y": 0.0, "z": 0.2 },
  "timestamp":   1714512100.4
}

World Forking

A fork is a full in-memory snapshot of the live world at a point in time — a sandbox for simulating moves before committing them. Forks are isolated: writes inside a fork never affect the live world, and live mutations don't propagate back into the fork.

async with wm.fork(ttl=120) as sim:
    # sim is a ForkClient — same interface as live client but isolated

    # Test a rearrangement without touching the real world
    await sim.move(table_id, position=(3, 0, 0))
    await sim.place("chair", "extra_chair", position=(2, 0, 0))

    # Query the simulated state
    nearby = await sim.query_spatial(center=(0, 0, 0), radius=5)
    state  = await sim.state()

    if layout_looks_good(nearby):
        # Commit: replay the moves on the live world
        await wm.move(table_id, position=(3, 0, 0))

# Fork is automatically deleted on context exit

Webhooks

Register an HTTP endpoint to receive push notifications when scene events occur. Useful for non-Python agents, external dashboards, or any consumer that can't maintain a gRPC stream.

# Register — receive all events
hook = await wm.register_webhook(
    url="https://my-agent.internal/universtato-events",
    events=["*"],          # or ["placed", "removed", "trigger:enter"]
    secret="my-secret",   # optional HMAC-SHA256 signing
)

# Later, unregister
await wm.unregister_webhook(hook["webhook_id"])

Event filter values

Payload envelope

{
  "event":     "placed",
  "timestamp": 1714512100.4,
  "payload":   { /* full SceneObject dict */ }
}

Signature verification

import hmac, hashlib

def verify(body: bytes, signature: str, secret: str) → bool:
    expected = "sha256=" + hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature)

# In your HTTP handler:
sig = request.headers.get("X-Universtato-Signature")
assert verify(request.body, sig, "my-secret")

Natural Language Interface

The NL interface accepts plain-English commands and translates them to scene mutations. Requires an Anthropic API key on the server (WORLDMODEL_API_KEY env var or --api-key flag).

# Via SDK
result = await wm.say("put a round oak table in the center of the room")
print(result["interpretation"])  # "Placing circular table at (0, 0, 0)"

# Via HTTP
POST /nl
{ "text": "move the sofa 2 meters to the left" }

→ { "success": true, "interpretation": "...", "command_type": "move" }

ROS 2 Bridge

The bridge connects Universtato to a live ROS 2 graph. Run it alongside the server on any machine with ROS 2 sourced.

# Install requirements
source /opt/ros/humble/setup.bash   # or iron / jazzy

# Start the bridge
python -m universtato.bridge.ros2 \
    --server http://localhost:8080 \
    --poll-hz 10 \
    --node-name universtato_bridge

Published topics (Universtato → ROS 2)

Subscribed topics (ROS 2 → Universtato)

Export

glTF 2.0

GET http://localhost:8080/export/gltf
→ universtato.gltf (self-contained JSON, no external buffers)

USD / USDC

GET http://localhost:8080/export/usd
→ universtato.usdc (binary USD, compatible with Omniverse / Blender)

# Requires usd-core:
pip install usd-core

Physics properties are stored as universtato:physics:* custom USD attributes. Affordances as universtato:affordances (string array). Metadata as universtato:metadata:*.

HTTP API Reference

Event Types

Events are delivered over the WebSocket stream (from /ws) and via gRPC SubscribeScene. The WebSocket sends individual events or batched arrays at up to 100Hz.

# Subscribe to all scene updates via gRPC
async def on_event(event_type: str, obj: dict | None):
    if event_type == "placed":
        print(f"New: {obj['name']}")
    elif event_type == "removed":
        print(f"Gone: {obj['id']}")

await wm.subscribe(on_event)