---
title: "vuer.rtc.types"
section: "Python API"
order: 20
description: "Python API reference for vuer.rtc.types"
---

# vuer.rtc.types

Core types for the vuer.rtc CRDT-based data store.

This module provides Python equivalents to the TypeScript @vuer-ai/vuer-rtc types,
enabling a consistent API across Python and JavaScript clients.

Key structures:
- VectorClock: Causal ordering via vector clocks
- Operation: Individual CRDT operations (node.insert, vector3.set, etc.)
- CRDTMessage: Message envelope containing batched operations
- SceneNode: Individual node in the scene graph
- SceneGraph: The computed state (map of nodes)
- JournalEntry: Committed message with ack/deleted status
- EditBuffer: Uncommitted operations awaiting commit
- Snapshot: Checkpoint for fast replay
- ClientState: Full client state container

## create_vector_clock

```python
def create_vector_clock() -> VectorClock
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L34)

Create an empty vector clock.

## increment_clock

```python
def increment_clock(clock: VectorClock, session_id: str) -> VectorClock
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L39)

Return a new clock with the session's counter incremented.

## merge_clocks

```python
def merge_clocks(clock1: VectorClock, clock2: VectorClock) -> VectorClock
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L46)

Merge two vector clocks, taking the max of each session's counter.

## compare_clocks

```python
def compare_clocks(clock1: VectorClock, clock2: VectorClock) -> Literal['before', 'after', 'concurrent', 'equal']
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L54)

Compare two vector clocks for causal ordering.
Returns:
  - "before" if clock1 happened before clock2
  - "after" if clock1 happened after clock2
  - "concurrent" if neither happened before the other
  - "equal" if they are identical

## OType

```python
class OType(str, Enum)
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L92)

All supported operation types (dtype.operation format).

```python
NUMBER_SET = 'number.set'
```

```python
NUMBER_ADD = 'number.add'
```

```python
NUMBER_MULTIPLY = 'number.multiply'
```

```python
NUMBER_MIN = 'number.min'
```

```python
NUMBER_MAX = 'number.max'
```

```python
STRING_SET = 'string.set'
```

```python
STRING_CONCAT = 'string.concat'
```

```python
TEXT_INIT = 'text.init'
```

```python
TEXT_INSERT = 'text.insert'
```

```python
TEXT_DELETE = 'text.delete'
```

```python
BOOLEAN_SET = 'boolean.set'
```

```python
BOOLEAN_OR = 'boolean.or'
```

```python
BOOLEAN_AND = 'boolean.and'
```

```python
VECTOR3_SET = 'vector3.set'
```

```python
VECTOR3_ADD = 'vector3.add'
```

```python
VECTOR3_MULTIPLY = 'vector3.multiply'
```

```python
VECTOR3_APPLY_EULER = 'vector3.applyEuler'
```

```python
VECTOR3_APPLY_QUATERNION = 'vector3.applyQuaternion'
```

```python
EULER_SET = 'euler.set'
```

```python
EULER_ADD = 'euler.add'
```

```python
QUATERNION_SET = 'quaternion.set'
```

```python
QUATERNION_MULTIPLY = 'quaternion.multiply'
```

```python
COLOR_SET = 'color.set'
```

```python
COLOR_BLEND = 'color.blend'
```

```python
ARRAY_SET = 'array.set'
```

```python
ARRAY_PUSH = 'array.push'
```

```python
ARRAY_UNION = 'array.union'
```

```python
ARRAY_REMOVE = 'array.remove'
```

```python
OBJECT_SET = 'object.set'
```

```python
OBJECT_MERGE = 'object.merge'
```

```python
NODE_INSERT = 'node.insert'
```

```python
NODE_REMOVE = 'node.remove'
```

```python
META_UNDO = 'meta.undo'
```

```python
META_REDO = 'meta.redo'
```

## Operation

```python
class Operation
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L161)

Base operation structure for all CRDT operations.

Attributes:
    key: Node key (e.g., 'cube-1', 'scene')
    otype: Operation type in dtype.operation format (e.g., 'vector3.add')
    path: Property path using dot notation (e.g., 'transform.position')
    value: Operation-specific value (type depends on otype)

```python
key: str
```

```python
otype: str
```

```python
path: str
```

```python
value: Any = None
```

```python
tag: Optional[str] = None
```

```python
parent_key: Optional[str] = None
```

```python
target_msg_id: Optional[str] = None
```

## Operation.to_dict

```python
def to_dict(self) -> Dict[str, Any]
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L184)

Serialize operation to dictionary.

## Operation.from_dict

```python
def from_dict(cls, data: Dict[str, Any]) -> 'Operation'
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L202)

Deserialize operation from dictionary.

## CRDTMessage

```python
class CRDTMessage
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L221)

Message envelope for batched CRDT operations.

Attributes:
    id: Unique message identifier (format: sessionId:sequence)
    session_id: Session that created this message
    clock: Vector clock for causal ordering
    lamport_time: Lamport timestamp for total ordering
    timestamp: Wall-clock time in milliseconds since epoch
    ops: Array of operations in this batch

```python
id: str
```

```python
session_id: str
```

```python
clock: VectorClock
```

```python
lamport_time: int
```

```python
timestamp: float
```

```python
ops: List[Operation]
```

## CRDTMessage.to_dict

```python
def to_dict(self) -> Dict[str, Any]
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L241)

Serialize message to dictionary.

## CRDTMessage.from_dict

```python
def from_dict(cls, data: Dict[str, Any]) -> 'CRDTMessage'
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L253)

Deserialize message from dictionary.

## generate_message_id

```python
def generate_message_id(session_id: str, sequence: int) -> str
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L265)

Generate a message ID in the format sessionId:sequence.

## generate_uuid

```python
def generate_uuid() -> str
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L270)

Generate a UUID string.

## SceneNode

```python
class SceneNode
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L281)

Individual node in the scene graph.

Attributes:
    key: Unique key (human-friendly identifier)
    tag: Node type (Scene, Mesh, Group, etc.)
    name: Display name
    children: List of child node keys
    clock: Vector clock when created
    lamport_time: Lamport timestamp
    created_at: Creation timestamp (ms since epoch)
    updated_at: Last update timestamp (ms since epoch)
    deleted_at: Soft delete marker (tombstone), None if not deleted
    properties: Dynamic properties stored by path

```python
key: str
```

```python
tag: str
```

```python
name: str = ''
```

```python
children: List[str] = field(default_factory=list)
```

```python
clock: VectorClock = field(default_factory=dict)
```

```python
lamport_time: int = 0
```

```python
created_at: float = field(default_factory=lambda: time.time() * 1000)
```

```python
updated_at: float = field(default_factory=lambda: time.time() * 1000)
```

```python
deleted_at: Optional[float] = None
```

```python
properties: Dict[str, Any] = field(default_factory=dict)
```

## SceneNode.get_property

```python
def get_property(self, path: str, default: Any=None) -> Any
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L309)

Get a property value by dot-notation path.

## SceneNode.set_property

```python
def set_property(self, path: str, value: Any) -> None
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L313)

Set a property value by dot-notation path.

## SceneNode.is_deleted

```python
def is_deleted(self) -> bool
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L318)

Check if this node has been soft-deleted.

## SceneNode.to_dict

```python
def to_dict(self) -> Dict[str, Any]
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L322)

Serialize node to dictionary.

## SceneNode.from_dict

```python
def from_dict(cls, data: Dict[str, Any]) -> 'SceneNode'
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L340)

Deserialize node from dictionary.

## SceneNode.copy

```python
def copy(self) -> 'SceneNode'
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L369)

Create a deep copy of this node.

## SceneGraph

```python
class SceneGraph
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L391)

The computed state of the scene as a flattened map of nodes.

Attributes:
    nodes: Dictionary mapping node keys to SceneNode instances
    root_key: Key of the root node (typically "scene")

```python
nodes: Dict[str, SceneNode] = field(default_factory=dict)
```

```python
root_key: str = 'scene'
```

## SceneGraph.get_node

```python
def get_node(self, key: str) -> Optional[SceneNode]
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L403)

Get a node by key.

## SceneGraph.set_node

```python
def set_node(self, node: SceneNode) -> None
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L407)

Add or update a node in the graph.

## SceneGraph.remove_node

```python
def remove_node(self, key: str) -> Optional[SceneNode]
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L411)

Remove a node from the graph. Returns the removed node or None.

## SceneGraph.has_node

```python
def has_node(self, key: str) -> bool
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L415)

Check if a node exists.

## SceneGraph.get_children

```python
def get_children(self, key: str) -> List[SceneNode]
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L419)

Get all child nodes of a given node.

## SceneGraph.to_dict

```python
def to_dict(self) -> Dict[str, Any]
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L426)

Serialize graph to dictionary.

## SceneGraph.from_dict

```python
def from_dict(cls, data: Dict[str, Any]) -> 'SceneGraph'
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L434)

Deserialize graph from dictionary.

## SceneGraph.copy

```python
def copy(self) -> 'SceneGraph'
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L441)

Create a deep copy of this graph.

## create_empty_graph

```python
def create_empty_graph(root_key: str='scene') -> SceneGraph
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L449)

Create an empty scene graph with a root node.

Args:
    root_key: Key for the root node (default: "scene")

Returns:
    A new SceneGraph with an empty root Scene node

## JournalEntry

```python
class JournalEntry
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L475)

A committed message with acknowledgment and deletion status.

Attributes:
    msg: The CRDT message
    ack: Whether the server has acknowledged this message
    deleted_at: Timestamp when this entry was undone (None if not undone)

```python
msg: CRDTMessage
```

```python
ack: bool = False
```

```python
deleted_at: Optional[float] = None
```

## JournalEntry.is_deleted

```python
def is_deleted(self) -> bool
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L489)

Check if this entry has been undone.

## JournalEntry.to_dict

```python
def to_dict(self) -> Dict[str, Any]
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L493)

Serialize entry to dictionary.

## JournalEntry.from_dict

```python
def from_dict(cls, data: Dict[str, Any]) -> 'JournalEntry'
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L504)

Deserialize entry from dictionary.

## EditBuffer

```python
class EditBuffer
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L519)

Buffer for uncommitted operations awaiting commit.

Attributes:
    ops: List of pending operations
    start_graph: Graph state when edits started (for cancel/revert)

```python
ops: List[Operation] = field(default_factory=list)
```

```python
start_graph: Optional[SceneGraph] = None
```

## EditBuffer.is_empty

```python
def is_empty(self) -> bool
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L531)

Check if there are no pending edits.

## EditBuffer.add

```python
def add(self, op: Operation) -> None
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L535)

Add an operation to the buffer.

## EditBuffer.clear

```python
def clear(self) -> List[Operation]
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L539)

Clear the buffer and return the operations.

## EditBuffer.to_dict

```python
def to_dict(self) -> Dict[str, Any]
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L546)

Serialize buffer to dictionary.

## EditBuffer.from_dict

```python
def from_dict(cls, data: Dict[str, Any]) -> 'EditBuffer'
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L556)

Deserialize buffer from dictionary.

## Snapshot

```python
class Snapshot
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L573)

Checkpoint for fast replay.

Attributes:
    graph: Scene graph state at this checkpoint
    vector_clock: Vector clock value at checkpoint
    lamport_time: Max lamport time baked into snapshot
    journal_index: Number of journal entries baked into snapshot

```python
graph: SceneGraph
```

```python
vector_clock: VectorClock
```

```python
lamport_time: int
```

```python
journal_index: int
```

## Snapshot.to_dict

```python
def to_dict(self) -> Dict[str, Any]
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L589)

Serialize snapshot to dictionary.

## Snapshot.from_dict

```python
def from_dict(cls, data: Dict[str, Any]) -> 'Snapshot'
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L599)

Deserialize snapshot from dictionary.

## create_initial_snapshot

```python
def create_initial_snapshot(graph: Optional[SceneGraph]=None) -> Snapshot
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L609)

Create an initial snapshot with an empty or provided graph.

## ClientState

```python
class ClientState
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L625)

Full client state container for the CRDT data store.

Attributes:
    session_id: Unique identifier for this client session
    graph: Current computed state
    journal: List of committed messages with ack status
    edits: Uncommitted operations awaiting commit
    snapshot: Checkpoint for fast replay
    lamport_time: Current Lamport timestamp
    vector_clock: Current vector clock

```python
session_id: str
```

```python
graph: SceneGraph
```

```python
journal: List[JournalEntry]
```

```python
edits: EditBuffer
```

```python
snapshot: Snapshot
```

```python
lamport_time: int
```

```python
vector_clock: VectorClock
```

## ClientState.to_dict

```python
def to_dict(self) -> Dict[str, Any]
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L647)

Serialize state to dictionary.

## ClientState.from_dict

```python
def from_dict(cls, data: Dict[str, Any]) -> 'ClientState'
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L660)

Deserialize state from dictionary.

## create_initial_state

```python
def create_initial_state(session_id: str, snapshot: Optional[Snapshot]=None) -> ClientState
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/rtc/types.py#L673)

Create an initial client state.

Args:
    session_id: Unique identifier for this session
    snapshot: Optional snapshot to restore from

Returns:
    A new ClientState initialized with the given session ID
