Tools

Tools are async Python handlers wrapped in a ToolSpec. The @define_tool(...) decorator builds the spec and binds the handler in one step.

from flowai_harness import define_tool, glimpse

@define_tool(
    name="search_products",
    description="Search products by query.",
    input_schema={"query": str, "limit": int},
    approval="never",
)
async def search_products(args, ctx):
    products = await ctx.product_catalog.search(args["query"], limit=args["limit"])
    return {
        "products": products,
        "glimpse": glimpse({
            "resultCount": len(products),
            "preview": [product["id"] for product in products[:3]],
        }),
    }

Decorator arguments

define_tool takes the following parameters:

input_schema accepts either a mapping of field name to Python type (a lightweight shorthand the harness normalizes to JSON Schema) or a full Pydantic model class:

Dict shorthand

from flowai_harness import define_tool

@define_tool(
    name="search_products",
    description="Search products by query.",
    input_schema={"query": str, "limit": int},
    approval="never",
)
async def search_products(args, ctx):
    ...

Pydantic model

from pydantic import BaseModel

from flowai_harness import define_tool

class SearchProductsInput(BaseModel):
    query: str
    limit: int = 10

@define_tool(
    name="search_products",
    description="Search products by query.",
    input_schema=SearchProductsInput,
    approval="never",
)
async def search_products(args, ctx):
    ...

Both forms normalize to the same JSON Schema on the wire.

Handler signature

Handlers are async and receive (args, ctx):

• args — a dict of input values produced by the model. input_schema constrains what the model is asked to produce and shapes the tool definition it sees, but the Rust runtime (flowai-runtime) does not re-validate arguments before dispatch — it forwards them to the handler as-is. Treat args as untrusted model output and validate inside the handler wherever shape or content matters.
• ctx — a ToolContext carrying runtime metadata and Python services bound with create_runtime(..., services=...). It supports mapping access (ctx["tool_use_id"]), .get(...), attribute access for valid service names (ctx.product_catalog), ctx.services, and the runtime reference client at ctx.references.

services is Python-owned and only reaches Python tool callbacks. Built-in Rust toolkits instead use data_environment — the runtime-side descriptor for catalog, KV, and target-database backends (see Data Environment) — so those handles are not injected into custom Python handlers unless your host application wraps them as services explicitly. Action dispatchers — the host callback the executor uses to enact approved plan actions — are also separate from tool service injection; see Using platform API tools and an action dispatcher together for the dispatcher-side dependency pattern. ctx.references is runtime-owned and writes to the same registry used by resolveRef, glimpseRef, and plan hydration.

The return value should be JSON-serializable. The optional "glimpse" field on the result is treated as a small summary the runtime can surface in subsequent prompts.

Tools that hand large or sensitive values to later turns ("pointer-producing" tools) should create typed references through ctx.references and return the {kind, id} handle plus a small "glimpse", instead of returning synthetic ids. The full resolveProductSet example and the resolveRef/glimpseRef decision rule live in References & Glimpses.

Built-in toolkits

Agents can also receive Rust-native toolkit ids through the toolkits argument. These tools are rendered into the agent prompt and dispatched by the runtime:

Documents and extracted knowledge are catalog entity kinds. Knowledge ingestion can project them into a writable catalog, and agents retrieve their typed catalog details through the catalog toolkit. See Knowledge and Documents for ingestion details.

Role constructors add their required built-in tools separately from toolkit selection. Passing toolkits=["catalog"] to define_planner(...) means "add catalog access" rather than "replace planner tools"; the planner still receives storePlan and getPlan. The same applies to executors, which keep getPlan, executePlan, and the reference helpers when additional toolkits are listed. Explicitly listing toolkits=["plans"] requests the full plan lifecycle toolkit.

MCP serving

Tools attached to a runtime specialist can also be exposed to MCP clients with flowai_harness.mcp. Python-defined callbacks stay in the Python process hosting the MCP server, while built-in toolkits use the Rust dependencies from data_environment. The host-side API is documented in Runtime; see Expose Tools Over MCP for stdio, Streamable HTTP, and CLI examples.

Approval policies

The approval argument controls when the runtime pauses for explicit approval before invoking the tool:

• "never" — invoke immediately. This is the default.
• "always" — emit an approval_required event and pause until the host calls runtime.respond_to_approval(...).
• A callable — register a dynamic predicate. The harness assigns a binding id (or uses binding_id if provided) and forwards it to the runtime; the predicate is called per invocation to decide whether approval is needed.
• A mapping — the explicit wire form: {"kind": "never"}, {"kind": "always"}, or {"kind": "dynamic", "value": "<binding-id>"} referencing an already-registered predicate.

from flowai_harness import define_tool

def approve_expensive_writes(args, ctx) -> bool:
    return args.get("amount", 0) > 10_000

@define_tool(
    name="post_journal_entry",
    description="Post an accounting journal entry.",
    input_schema={"account": str, "amount": float},
    approval=approve_expensive_writes,
)
async def post_journal_entry(args, ctx):
    ...

The approval value is metadata; the gate itself is enforced by the runtime — see Approval policies in Agents.

Attaching tools to agents

Tools attach to any agent role via the tools argument. define_runtime(...) collects them transitively so you do not need to pass the same tool to both the agent and the runtime spec:

from flowai_harness import define_planner, define_runtime, define_tenant

tenant = define_tenant("acme", "v1")

planner = define_planner(
    name="scenario_planner",
    model="claude-sonnet-4-6",
    plan=scenario_plan,
    tools=[search_products],
    prompt="...",
)

runtime_spec = define_runtime(
    tenant=tenant,
    agents=[coordinator, planner, executor, specialist],
)

See also

• define_tool reference
• References & Glimpses — building summaries with glimpse(...).
• Runtime — attaching Python services and Rust data-environment dependencies.