Tool#

OpenRath separates model-callable capabilities from the side effects that actually happen in a sandbox. FlowToolCall faces the model and session loop; BackendTool* payloads face the backend and sandbox.

This page explains how model tool calls enter OpenRath, the schema/runtime contract for FlowToolCall, how backend payloads execute in local/OpenSandbox, and how results are written back to Session.

The diagram below follows one tool from definition to model schema, backend execution, and final tool_result writeback.

Tool system overview

FlowToolCall has two faces: a schema that the model can call and a Python callable that can run directly or dispatch work to a backend.#

Overview#

A tool call crosses three boundaries:

Layer

Object

Responsibility

Model interface

JSON schema

Tells the model which tools exist and what arguments they accept.

Python runtime

FlowToolCall

Receives the current Session and model arguments, then runs Python logic.

Sandbox backend

BackendTool* payload

Runs command, file, and code operations in local or OpenSandbox.

This split gives custom tools two implementation paths: simple tools can return a dict directly from the Python process; tools with side effects can hand work to the backend through session.require_sandbox().dispatch(...).

Source map#

File

Responsibility

src/rath/flow/tool/base.py

FlowToolCall abstract base class.

src/rath/flow/tool/system_tool.py

Built-in tools and backend payload factories.

src/rath/flow/tool/tool_table.py

Tool merge, schema conversion, name conflict handling.

src/rath/session/loop.py

Tool call dispatch and tool result serialization.

src/rath/backend/tool_types.py

Backend-facing payload dataclasses.

src/rath/backend/results.py

Backend result dataclasses.

FlowToolCall Contract#

FlowToolCall is the tool interface recognized by the session loop. It must provide name, parameters, and __call__.

from collections.abc import Mapping
from typing import Any

from rath.flow.tool import FlowToolCall
from rath.session import Session


class AddOneTool(FlowToolCall):
    @property
    def name(self) -> str:
        return "add_one"

    @property
    def description(self) -> str | None:
        return "Add 1 to x"

    @property
    def parameters(self) -> Mapping[str, Any]:
        return {
            "type": "object",
            "properties": {"x": {"type": "integer"}},
            "required": ["x"],
            "additionalProperties": False,
        }

    def __call__(self, session: Session, arguments: Mapping[str, Any]) -> int:
        return int(arguments["x"]) + 1

Member

Purpose

name

Name of the LLM function tool and the key used by the loop to find the Python tool object.

description

Optional description to help the model decide when to call it.

parameters

JSON Schema object placed in the OpenAI-compatible tools field.

__call__(session, arguments)

Executes the tool. session is the current output session; arguments are JSON arguments generated by the model.

__call__ can return ordinary Python objects. The session loop serializes them to a JSON string and writes a tool_result chunk.

How The Tool Table Is Built#

Each run_session_loop(...) call builds a tool table first.

Step

Behavior

1

global_system_tools() returns built-in tools.

2

merge_tools_for_loop(user_tools) adds user tools to the table.

3

If a user tool name overrides a built-in tool name, raises ToolNameConflictError.

4

tools_dict_to_schemas(table) generates OpenAI-style function schemas.

There are currently two built-in tools:

Tool name

Behavior

Backend payload

run_shell_command

Runs one shell command in the active sandbox workspace.

BackendToolCommandRun

write_workspace_file

Writes a UTF-8 text file into the sandbox workspace.

BackendToolFilesWrite

The built-in shell tool has two guards: it rejects multiline commands and commands longer than 2048 characters. This is the minimum protection in the current implementation; the concrete security boundary is set by the backend.

From Model Response To Tool Result#

After the model returns a tool call, run_session_loop(...) follows this path:

assistant tool_calls
  -> table[tool_name]
  -> executor.dispatch_tool(out, flow_tool, arguments)
  -> flow_tool(out, arguments)
  -> Python result or backend dispatch result
  -> tool_feedback_chunk(...)
  -> out.chunk_table

The important value is out. The loop creates the output session first and moves the input user session’s sandbox to out. Therefore tools receive the output session, and both side effects and tool results belong to that loop output.

def __call__(self, session: Session, arguments):
    sandbox = session.require_sandbox()
    return sandbox.dispatch(...)

Tool execution results are converted to strings by _summarize_dispatch_result(...). Common mappings are:

Return value

Content written to tool_result

Plain dict/list/int/str

JSON text; truncated above 48,000 chars.

Pydantic model

JSON after model_dump(mode="json").

CommandResult

exit_code, stdout, stderr, elapsed_ms.

FileContent

data, truncated above 12,000 chars.

FileEntries

Up to the first 500 entries.

FileWriteResult

bytes_written.

CodeResult

text, stdout, stderr, error.

ToolExecutionFailure

ok=false, error_kind, message, detail.

bool

{"ok": true} or {"ok": false}.

BackendTool payload#

Backend payloads are lower-level descriptions of sandbox operations. They are not exposed directly to the model; usually a FlowToolCall builds them and passes them to the sandbox.

Payload

Fields

Purpose

BackendToolCommandRun

cmd, env, cwd, stdin, timeout

Runs a shell command.

BackendToolFilesRead

path, encoding

Reads a file.

BackendToolFilesWrite

path, data, mode

Writes a file.

BackendToolFilesList

path

Lists a directory.

BackendToolFilesExists

path

Checks whether a path exists and returns bool.

BackendToolCodeRun

code, language, timeout

Runs a code snippet.

Example: a custom tool can convert model arguments into a backend payload.

from rath.backend import BackendToolFilesRead
from rath.flow.tool import FlowToolCall


class ReadTextTool(FlowToolCall):
    @property
    def name(self):
        return "read_text"

    @property
    def parameters(self):
        return {
            "type": "object",
            "properties": {"path": {"type": "string"}},
            "required": ["path"],
            "additionalProperties": False,
        }

    def __call__(self, session, arguments):
        call = BackendToolFilesRead(path=str(arguments["path"]))
        return session.require_sandbox().dispatch(call)

When To Return Python Objects Directly#

Returning Python objects directly fits tools without external side effects, such as string processing, argument validation, lightweight computation, or reading in-process caches.

def __call__(self, session, arguments):
    text = str(arguments["text"])
    return {"words": len(text.split())}

This kind of tool does not need a sandbox and does not trigger the backend lifecycle.

When To Use Backend Dispatch#

Backend dispatch fits tools that need side effects or an isolated environment, such as reading or writing the workspace, running shell commands, executing code, or accessing dependencies inside a container.

def __call__(self, session, arguments):
    sandbox = session.require_sandbox()
    return sandbox.dispatch(...)

This kind of tool requires the user session entering the loop to have a sandbox target or open handle. Otherwise require_sandbox() raises RuntimeError.

Where Stream API Fits#

BackendSandbox.stream() is the concurrency API at the backend layer. It is separate from the model tool-call loop.

Scenario

Current behavior

session loop handling model-returned tool calls

Calls executor dispatch one by one.

manually submitted backend payloads

Can create multiple streams.

same stream

FIFO queue, one worker thread, one operation at a time.

different streams

Different worker threads can make progress concurrently.

stream event

record_event() and wait_event(...) can create ordering across streams.

Typical FlowToolCall implementations do not need to touch streams. Use them when manually coordinating backend-level concurrent payloads.

Edge Cases#

Behavior

Current implementation

User tool name overrides a built-in tool

merge_tools_for_loop(...) raises ToolNameConflictError.

Model returns unparseable JSON arguments

Loop writes invalid_tool_arguments JSON error.

Model requests an unknown tool

Loop writes unknown_tool JSON error.

Tool execution raises an exception

Loop catches it and writes tool_execution_exception JSON error.

Plain return value is too long

JSON text is truncated above 48,000 chars.

FileContent is too long

Text is truncated above 12,000 chars.

Too many FileEntries

Serializes only the first 500 entries.

Shell command is multiline or too long

Built-in shell tool raises ValueError; loop converts it to a tool execution error.

Code Reading Checkpoints#

Question

Where to look

Minimal FlowToolCall interface

src/rath/flow/tool/base.py

Built-in tool implementation

src/rath/flow/tool/system_tool.py

Name conflicts and schema ordering

src/rath/flow/tool/tool_table.py

How tool results are written back to session

src/rath/session/loop.py

Backend payload types

src/rath/backend/tool_types.py

Backend result types

src/rath/backend/results.py

Test Coverage#

Behavior

Tests

tool factory/result basics

tests/unit/test_flow_tool.py, tests/unit/test_calls.py, tests/unit/test_results.py

tool merge conflict

tests/session/test_tool_registry.py

custom FlowToolCall loop result

tests/flow/test_flow_tool_user_subclass.py

loop edge cases

tests/session/test_run_session_loop_edges.py

workflow agent tools

tests/flow/test_workflow_agent.py