rath.flow.tool

Model-visible tool interface, built-in tools, tool table merging, and backend payload factories.

Source

Module	Source
rath.flow.tool.base	src/rath/flow/tool/base.py
rath.flow.tool.system_tool	src/rath/flow/tool/system_tool.py
rath.flow.tool.tool_table	src/rath/flow/tool/tool_table.py
rath.flow.tool.mcp_adapter	src/rath/flow/tool/mcp_adapter.py

Public contract

FlowToolCall

Member	Type	Description
name	str	OpenAI function tool name.
description	str | None	Optional tool description.
parameters	Mapping[str, Any]	JSON Schema object.
parallel_safe	bool	Whether the default async scheduler may run this tool beside another call of the same tool name.
__call__(session, arguments)	Any	Runtime execution entrypoint.
resource_key(arguments)	tuple	Concurrency grouping key. Calls with the same key run serially; distinct keys may run concurrently.

By default, tools are conservative: parallel_safe=False groups calls under ("global",). Tools that are safe to run concurrently can set parallel_safe=True or override resource_key(...) to group by file path, remote resource id, account id, or another domain-specific lock key.

Built-in tools

Tool	Name	Arguments	Behavior
FlowToolCommandRun	run_shell_command	{cmd: string}	Calls BackendToolCommandRun.
FlowToolFilesWrite	write_workspace_file	{path: string, content: string}	Calls BackendToolFilesWrite.

FlowToolCommandRun rejects multiline commands and commands longer than 2048 characters. FlowToolFilesWrite requires content to be text.

Tool table

Function	Returns	Behavior
global_system_tools()	dict[str, FlowToolCall]	Returns the in-process singleton built-in tool table.
merge_tools_for_loop(user_tools)	dict[str, FlowToolCall]	Merges built-in tools and user tools.
tools_dict_to_schemas(table)	tuple[RathLLMFunctionTool, ...]	Converts to OpenAI-style function tool schemas.

When a user tool name conflicts with a built-in tool name, merge_tools_for_loop(...) raises ToolNameConflictError.

MCP stdio adapter

The MCP adapter discovers stdio server tools and exposes each one as a normal FlowToolCall for the OpenRath session loop.

API	Returns	Behavior
MCPClient(command, args=None, env=None)	client	Synchronous wrapper around a stdio MCP server.
MCPClient.list_tools()	list	Opens a subprocess, initializes MCP, and returns raw tool definitions.
MCPClient.call_tool(name, arguments)	raw result	Opens a subprocess and calls one MCP tool.
MCPToolCall	FlowToolCall	Wraps one MCP tool as a model-visible OpenRath tool.
mcp_tools_from_server(command, args=None, env=None)	tuple	Discovers all tools from a stdio server.
mcp_tools_from_config(name=None)	tuple	Reads one MCP server entry from ~/.openrath/config.json.

The adapter currently supports stdio transport only. Each list/call opens a fresh subprocess on a shared dedicated event loop, so tool lifecycle is simple and there is no explicit close step.

Backend tool factories

Function	Returns
flow_tool_command_run(cmd, env=None, cwd=None, stdin=None, timeout=None)	BackendToolCommandRun
flow_tool_files_read(path, encoding="utf-8")	BackendToolFilesRead
flow_tool_files_write(path, data, mode=0o644)	BackendToolFilesWrite
flow_tool_files_list(path)	BackendToolFilesList
flow_tool_files_exists(path)	BackendToolFilesExists
flow_tool_code_run(code, language="python", timeout=None)	BackendToolCodeRun

Autodoc

class rath.flow.tool.FlowToolCall

User- or system-defined tool for the session loop (distinct from BackendTool).

Concurrency contract (used by the async session loop):

• parallel_safe: class attribute. True means the runtime may await this tool concurrently with other parallel_safe tools sharing a different resource_key(). False means the runtime must run this tool serially with respect to every other tool in the same round. Built-in exec/code tools default to False; filesystem reads and writes default to True; user tools default to False and must opt in explicitly.

• resource_key(): keys the runtime uses to serialize tools that touch the same resource. Same key → serial; different key → parallel. Default returns ("global",) for non-parallel-safe tools so they pile up on one queue; ("safe", name) for parallel-safe tools so they fan out freely.

abstract property parameters: Mapping[str, Any]

JSON Schema object for OpenAI parameters.

resource_key(arguments: Mapping[str, Any]) → tuple[str, ...]

Return the resource-key the async runtime serializes on.

Override for fs/exec tools to expose a meaningful key (path, sandbox handle, …). Default partitions tools into one global serial lane for non-parallel-safe tools and a per-name lane for parallel-safe tools.

class rath.flow.tool.FlowToolCommandRun

Built-in LLM tool: run one shell command in the active sandbox.

resource_key(arguments: Mapping[str, Any]) → tuple[str, ...]

Return the resource-key the async runtime serializes on.

Override for fs/exec tools to expose a meaningful key (path, sandbox handle, …). Default partitions tools into one global serial lane for non-parallel-safe tools and a per-name lane for parallel-safe tools.

property parameters: Mapping[str, Any]

JSON Schema object for OpenAI parameters.

class rath.flow.tool.FlowToolFilesWrite

Built-in LLM tool: write UTF-8 text into the sandbox workspace.

resource_key(arguments: Mapping[str, Any]) → tuple[str, ...]

Return the resource-key the async runtime serializes on.

Override for fs/exec tools to expose a meaningful key (path, sandbox handle, …). Default partitions tools into one global serial lane for non-parallel-safe tools and a per-name lane for parallel-safe tools.

property parameters: Mapping[str, Any]

JSON Schema object for OpenAI parameters.

rath.flow.tool.global_system_tools() → Mapping[str, FlowToolCall]

Process-wide built-in tools (singleton instances per name, immutable view).

Returns a types.MappingProxyType over the internal registry so callers cannot accidentally mutate the global state via the returned object. Callers that need a mutable working copy can still do dict(global_system_tools()); that explicit copy is local and does not touch the underlying registry.

rath.flow.tool.merge_tools_for_loop(user_tools: list[FlowToolCall] | None) → dict[str, FlowToolCall]

Merge user tools with global_system_tools().

Built-in names cannot be shadowed; duplicates raise ToolNameConflictError.

rath.flow.tool.tools_dict_to_schemas(table: Mapping[str, FlowToolCall]) → tuple[RathLLMFunctionTool, ...]

Convert a name-to-tool map into sorted OpenAI-style function specs.

rath.flow.tool.flow_tool_command_run(session: Session, cmd: str | Sequence[str], *, env: Mapping[str, str] | None = None, cwd: str | None = None, stdin: bytes | None = None, timeout: float | None = None) → Any

rath.flow.tool.flow_tool_files_read(session: Session, path: str, *, encoding: str | None = 'utf-8') → Any

rath.flow.tool.flow_tool_files_write(session: Session, path: str, data: bytes | str, *, mode: int = 420) → Any

rath.flow.tool.flow_tool_files_list(session: Session, path: str) → Any

rath.flow.tool.flow_tool_files_exists(session: Session, path: str) → Any

rath.flow.tool.flow_tool_code_run(session: Session, code: str, *, language: str = 'python', timeout: float | None = None) → Any

exception rath.flow.tool.ToolNameConflictError

Raised when a user tool name collides with a built-in system tool.

MCP adapter APIs are documented manually above because importing the third-party mcp package during a docs build can depend on the installed Pydantic minor version. Runtime users should import them from rath.flow.tool.mcp_adapter.

← API Reference