orxtra v0.13.0 /worker.src.orxtra.worker._native
Updated
On this page

NativeWorker connects to a brain over WebSocket and executes tool calls against the local filesystem via build_worker_tools (file, git, and exec tools), reconnecting with exponential backoff.

#worker.src.orxtra.worker._native

#worker.src.orxtra.worker._native

Native worker: standalone process that executes tools locally.

Connects to a brain over WebSocket, receives tool calls, executes them against the local filesystem, and returns results.

#_make_worker_exec_tool

python
def _make_worker_exec_tool(executable: str, description: str, read_root: Path) -> Tool

Build a simple exec-style tool for the worker.

Uses run_subprocess directly. The tool schema matches the historic exec tool: args (optional list[str]) and timeout (optional int).

#build_worker_tools

python
def build_worker_tools(root: Path, queue: WriteQueue, tracker: StaleWriteTracker, session_id: str) -> dict[str, Tool]

Build the essential tool set for a worker.

Constructs read, write, edit, grep, glob, stat, diff, list_dir, git, exec (pytest, uv), mkdir, move, copy, delete, multi_edit, and set_executable tools scoped to the given root.

#_serialize_message

python
def _serialize_message(msg_type: str, model: Any) -> str

Serialize a protocol message as a JSON envelope.

#NativeWorker

Standalone worker process that connects to a brain over WebSocket.

Receives ExecuteToolCall messages, executes tools locally against the given root directory, and returns ToolCallResult messages.

#run

python
async def run(self) -> None

Main loop: connect, register, process messages, reconnect on failure.

#_connect_and_run

python
async def _connect_and_run(self) -> None

Connect to the brain, register, and enter the message loop.

#_handle_message

python
async def _handle_message(self, ws: Any, raw: str) -> None

Parse and dispatch a single message from the brain.

#_execute_tool

python
async def _execute_tool(self, call: ExecuteToolCall) -> ToolCallResult

Execute a tool call locally and return the result.

#stop

python
def stop(self) -> None

Signal the worker to stop after the current message.

Search