# 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
```python
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
```python
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
```python
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
```python
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__`:

```python
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.