Custom FlowToolCall

To connect external capabilities to OpenRath, provide two pieces: a JSON Schema visible to the model, and a Python callable executable at runtime. This page uses WordCountTool to show the smallest useful FlowToolCall implementation.

Coverage

Topic	Result
Tool schema	How name, description, and parameters are exposed to the model.
Argument validation	Use Pydantic to validate JSON arguments generated by the model.
Runtime execution	__call__(session, arguments) is the tool execution entry point.
Result serialization	Regular Python return values are serialized by the loop into tool_result.
Sandbox access	Tools can call the Backend through session.require_sandbox().

Step 1: Define the Input Schema

from pydantic import BaseModel, Field


class WordCountInput(BaseModel):
    text: str = Field(description="Text to count.")

Key lines:

Line	Explanation
BaseModel	Describes tool input with a structured type.
Field(description=...)	Helps the model understand what the argument means.
model_json_schema()	Later becomes the parameters field in the LLM tool schema.

Step 2: Subclass FlowToolCall

from collections.abc import Mapping
from typing import Any

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


class WordCountTool(FlowToolCall):
    @property
    def name(self) -> str:
        return "word_count"

    @property
    def description(self) -> str:
        return "Count words in a text string."

    @property
    def parameters(self) -> Mapping[str, Any]:
        return WordCountInput.model_json_schema()

    def __call__(
        self,
        session: Session,
        arguments: Mapping[str, Any],
    ) -> dict[str, int]:
        model = WordCountInput.model_validate(dict(arguments or {}))
        return {"words": len(model.text.split())}

Key lines:

Line	Explanation
name	The model uses this name when returning a tool call.
description	Tells the model when to call the tool.
parameters	JSON Schema that determines which arguments the model should generate.
model_validate(...)	Validates the model-generated dict into a Python object.
return {"words": ...}	The loop serializes the dict as JSON text and writes it to tool_result.

__call__ receives the current Session. That means the tool can read Session state and can also call file, command, or code payloads through session.require_sandbox().dispatch(...).

Step 3: Pass It into the Session Loop

from rath import flow
from rath.session import Session, run_session_loop

tool = WordCountTool()
agent_session = Session.from_agent_prompt(
    "Call word_count before answering word-count questions."
)
user_session = Session.from_user_message(
    "Count the words in: OpenRath keeps agent state explicit."
).to("local")

out = run_session_loop(
    user_session=user_session,
    agent_session=agent_session,
    agent_provider=flow.Provider(api_key="sk-...", model="gpt-5.5"),
    tools=[tool],
    executor=scripted_executor,
)

Key lines:

Line	Explanation
tools=[tool]	Gives the custom tool to the loop.
.to("local")	Makes the user Session’s execution location explicit in this tutorial.
executor=scripted_executor	Keeps model responses fixed in tutorials or tests. Omit it in real runs.

run_session_loop(...) merges built-in tools with passed-in tools. If a tool name conflicts with a built-in tool name, it raises ToolNameConflictError.

Step 4: Read the Result

for row in out.chunk_table.rows:
    if row.kind.value == "tool_result":
        print(row.payload["name"], row.payload["content"])

Observed behavior:

• row.payload["name"] is word_count.

• row.payload["content"] is JSON text, for example {"words": 5}.

• If the tool raises an exception, the loop wraps the error as a tool failure payload visible to the model.

Step 5: Let the Tool Access the Sandbox

If the tool needs to read or write the workspace, get the sandbox inside __call__:

from rath.backend import BackendToolFilesRead

def __call__(self, session: Session, arguments: Mapping[str, Any]) -> dict[str, int]:
    model = WordCountInput.model_validate(dict(arguments or {}))
    sandbox = session.require_sandbox()
    content = sandbox.dispatch(
        BackendToolFilesRead(path=model.text, encoding="utf-8")
    )
    text = str(content.data)
    return {"words": len(text.split())}

This kind of tool requires the user Session to already be bound to a Backend. Otherwise, session.require_sandbox() fails.

Troubleshooting

Symptom	Check
Model does not call the tool	Strengthen the tool description or system prompt.
Argument validation fails	Check the Pydantic error and confirm parameters matches the prompt.
Tool name conflict	Use a unique name that does not overlap with built-in tool names.
Tool needs files but cannot find a sandbox	Confirm the user Session has called .to("local") or .to("opensandbox").
Return value is not JSON serializable	Return a dict, list, str, int, or Pydantic model.

Exercises

1. Add a lowercase: bool parameter to WordCountInput.

2. Return characters and lines from the tool as well.

3. Change the tool so it reads a workspace file and counts words in the file content.

Summary

• FlowToolCall provides both the tool schema and the Python execution logic.

• parameters is the JSON Schema visible to the model.

• __call__ is the runtime execution entry point.

• tools=[tool] is how a custom tool is passed to the agent.