> For the complete documentation index, see [llms.txt](/llms.txt). Every page on this site is also available as markdown at `<path>.md`.

# Approvals

The Flow AI runtime can pause mid-stream to ask the host application to approve a
proposed plan or a pending tool call. Python supplies the policy and the
response; the actual gate lives in the Rust runtime.

When `define_coordinator(..., approval={"plans": "always", "tools": "never"})`
is set, that policy becomes the runtime approval floor unless
`define_runtime(..., approval_policies=...)` supplies one explicitly. Individual
agents can override the floor with `approval={...}`, and individual tools under
an agent can override it with `tool_approvals={...}` or `define_tool(...,
approval=...)`. If the resolved policy requires approval, the runtime emits an
`approval-required` event, suspends the stream, and waits for
`runtime.respond_to_approval(...)` before continuing.

## Policy values

Runtime and agent approval policies accept the same shape for both channels:

| Value | Effect |
| --- | --- |
| `"never"` | Calls run without a gate. |
| `"always"` | Every call emits `approval-required` and waits. |
| `"default"` | Per-agent `approval={...}` only. Leave this channel unchanged from the runtime floor. |
| `{"kind": "dynamic", "value": "<predicate-id>"}` | Tool-only. The runtime calls a registered Python predicate per call. |

Plan approval supports `"never"` and `"always"`. Tool approval also supports
dynamic predicates, registered through `define_tool(..., approval=fn)` or by
passing the `approval_predicates` mapping to `create_runtime(...)`.

`"default"` is a per-agent sentinel, not a runtime-level value. In
`define_coordinator(..., approval={...})` (and the other `define_*` agent
helpers) a `"default"` channel is stripped before the spec is built, so the
runtime floor applies unchanged. Passing it to
`define_runtime(approval_policies={"plans": "default"})` raises a validation
error: at the runtime level an approval rule must be `"never"`, `"always"`, or
a dynamic rule.

```python
coordinator = define_coordinator(
    name="scenario_coordinator",
    model="claude-sonnet-4-6",
    routes=["scenario_planner", "scenario_executor"],
    approval={"plans": "always", "tools": "never"},
    prompt=coordinator_prompt,
)

executor = define_executor(
    name="scenario_executor",
    model="claude-sonnet-4-6",
    plan=scenario_plan,
    approval={"plans": "never", "tools": "never"},
    tool_approvals={"execute_query": "always"},
    prompt=executor_prompt,
)
```

Precedence is runtime floor, then agent override, then tool override. Tool
overrides are scoped by agent, so the same tool name can be approval-gated for
one specialist and ungated for another.

## The `approval-required` event

The stream emits a wire-shape dict. The `data` field carries the Rust
`ApprovalRequest` payload in camelCase:

```python
{
    "type": "approval-required",
    "data": {
        "id": "apr-1234",
        "kind": "plan",                # or "tool"
        "target": "demo-plan-1",       # plan id, or the tool name
        "payload": {...},              # plan body, or tool args
        "glimpse": {...},              # optional preview, may be absent
        "resourceId": "acme",
        "threadId": "thread-1",
        "correlationId": "..."         # optional, ties retries to the originating call
    }
}
```

For tool approvals, `target` is the tool name and `correlationId` is the LLM's
`tool_use_id`. For plan approvals, `target` and `correlationId` are both the
plan id.

## Responding to an approval

`runtime.respond_to_approval(approval_id, outcome, feedback=None, partial=None)`
is an async method:

```python
await runtime.respond_to_approval(
    event["data"]["id"],
    "approve",
    feedback="approved by smoke test",
)
```

Outcomes:

- `"approve"` — runs the gated action.
- `"reject"` — drops the action. Accept an optional `feedback` string.
- `"revise"` — plan only. The runtime re-invokes the planner with `partial`,
  which is forwarded as the `partial` payload the executor's `executePlan`
  tool returns. Tool approvals collapse `revise` to `reject`.

After the response lands, the runtime emits `approval-decision` and then the
gated event sequence — for a plan, the executor's tool calls; for a tool, the
`tool-invocation` result event.

## End-to-end example

The script below is a complete, runnable version of the same coordinator →
planner → executor flow the runtime's integration tests exercise. It uses the
scripted interpreter (`interpreter="scripted"`), so each `call_agent` prompt is
replayed deterministically instead of being sent to a live model: the planner
prompt is the `storePlan` tool call as JSON, and the executor prompt is the
matching `executePlan` call. The coordinator gates the plan; the host approves
it; only then does the executor's action dispatch run.

```python
import asyncio
import json

from flowai_harness import (
    create_runtime,
    define_coordinator,
    define_executor,
    define_tenant,
    define_plan,
    define_planner,
    define_runtime,
)

scenario_plan = define_plan(
    "DemoPlan",
    {
        "type": "object",
        "required": ["rationale", "actions"],
        "properties": {
            "rationale": {"type": "string"},
            "actions": {
                "type": "array",
                "minItems": 1,
                "items": {
                    "type": "object",
                    "required": ["kind"],
                    "properties": {
                        "kind": {"type": "string"},
                        "message": {"type": "string"},
                    },
                },
            },
        },
    },
)

coordinator = define_coordinator(
    "coordinator",
    model="claude-sonnet-4-6",
    prompt="Coordinate by delegating to the planner and executor.",
    routes=["planner", "executor"],
    approval={"plans": "always", "tools": "never"},
)
planner = define_planner(
    "planner",
    model="claude-sonnet-4-6",
    prompt="Store exactly one typed plan.",
    plan=scenario_plan,
)
executor = define_executor(
    "executor",
    model="claude-sonnet-4-6",
    prompt="Execute the requested plan.",
    plan=scenario_plan,
)

def dispatch_actions(actions, ctx):
    print(f"dispatching {len(actions)} approved action(s)")
    return {
        "entitiesAffected": len(actions),
        "summary": f"executed {len(actions)} action(s)",
        "details": None,
    }

runtime = create_runtime(
    define_runtime(
        tenant=define_tenant("acme", "v1"),
        agents=[coordinator, planner, executor],
        providers={"anthropic": {"apiKey": "unused"}},
    ),
    action_dispatcher=dispatch_actions,
    interpreter="scripted",
)

# The scripted interpreter replays these prompts instead of calling a model.
# The planner prompt is the storePlan tool call; the executor prompt is the
# executePlan call for the same plan id.
planner_prompt = json.dumps({
    "tool": "storePlan",
    "args": {
        "specName": "DemoPlan",
        "planId": "demo-plan-1",
        "body": {
            "rationale": "deterministic demo plan",
            "actions": [
                {"kind": "record_counter", "message": "record the approved action"},
            ],
        },
    },
})
executor_prompt = json.dumps({
    "tool": "executePlan",
    "args": {"planId": "demo-plan-1"},
})

async def main():
    stream = runtime.query(
        json.dumps({"script": [
            {"tool": "call_agent", "args": {"agent": "planner", "prompt": planner_prompt}},
            {"tool": "call_agent", "args": {"agent": "executor", "prompt": executor_prompt}},
        ]}),
        thread_id="thread-1",
    )
    async for event in stream:
        if event["type"] == "approval-required":
            data = event["data"]
            print(f"approval-required: kind={data['kind']} target={data['target']}")
            await runtime.respond_to_approval(
                data["id"],
                "approve",
                feedback="approved by host",
            )
        elif event["type"] == "approval-decision":
            print(f"approval-decision: {event['data']['outcome']}")

asyncio.run(main())
```

Running the script prints:

```text
approval-required: kind=plan target=demo-plan-1
dispatching 1 approved action(s)
approval-decision: {'outcome': 'approve'}
```

The gate fires when the executor calls `executePlan`: the stream suspends on
`approval-required`, `dispatch_actions` runs only after the `"approve"`
response lands, and the `approval-decision` event then records the outcome
before the executor's result events complete the stream.

<Callout type="info" title="Note">

The coordinator must not dispatch any side effect before the gate resolves.
The runtime guarantees this — `action_dispatcher` is only invoked after an
`"approve"` response lands. Use the `interpreter="scripted"` mode in tests
to drive the exact sequence above without a live model.

</Callout>

## Action dispatcher contract

`action_dispatcher` is the host adapter that applies approved plan actions to
your platform. It receives the full action list and a runtime-built context:

```python
def dispatch_actions(actions, ctx):
    ...
```

`actions` is a JSON-serializable list of normalized harness actions. Flat plan
actions are converted to the canonical shape before dispatch:

```python
[
    {
        "kind": "price_change",
        "payload": {"product_id": "sku-1", "new_price": 12.5},
        "references": [],
    }
]
```

`ctx["resolved_refs"]` contains any action references hydrated by the runtime,
grouped by reference kind and id:

```python
{
    "ProductSet": {
        "ref-123": {"product_ids": ["sku-1", "sku-2"]}
    }
}
```

The dispatcher may return `None`, which is treated as an empty execution result.
Otherwise it must return an object matching the execution-result envelope:

```python
return {
    "entitiesAffected": len(actions),  # required non-negative integer
    "summary": "Updated 3 products",   # optional string or None
    "details": {                       # optional JSON value
        "platformJobId": "job-123",
    },
}
```

Only `entitiesAffected`, `summary`, and `details` are accepted at the top level.
Put domain-specific metadata under `details`. Invalid dispatcher returns fail
`executePlan`, and the plan is marked failed rather than silently discarding
malformed data.

## Using platform API tools and an action dispatcher together

Most production agents use custom tools and an action dispatcher against the
same external platform API. The split is:

- Custom tools let the planner inspect, search, simulate, or preview platform
  state while it is building a plan.
- The action dispatcher performs the approved writes by translating plan actions
  into platform API calls.

For example, a commerce platform might expose a generic action API:

```python
class MockCommerceApi:
    def preview_price_change(self, product_id, new_price):
        return {"productId": product_id, "newPrice": new_price, "marginOk": True}

    def preview_availability_change(self, product_id, available):
        return {"productId": product_id, "available": available, "safeToApply": True}

    def create_action(self, *, action_type, payload):
        return {
            "id": f"action-{action_type.lower()}-{payload['productId']}",
            "type": action_type,
            "payload": payload,
        }
```

The plan still uses typed action variants. The action `kind` is the stable
harness-side discriminator; the dispatcher maps it to the platform's API type.

```python
from typing import Literal

from pydantic import BaseModel

from flowai_harness import TaggedUnion, define_plan, define_tool

class PriceChange(BaseModel):
    kind: Literal["price_change"]
    product_id: str
    new_price: float

class AvailabilityChange(BaseModel):
    kind: Literal["availability_change"]
    product_id: str
    available: bool

CommercialAction = TaggedUnion(PriceChange, AvailabilityChange)

class CommercialPlan(BaseModel):
    rationale: str
    actions: list[CommercialAction]

commercial_plan = define_plan("CommercialPlan", CommercialPlan)
```

Tools can use the platform client for non-final operations. Pass the service to
`create_runtime(..., services=...)`, then read it from the tool context:

```python
api = MockCommerceApi()

@define_tool(
    "preview_price_change",
    {"product_id": str, "new_price": float},
    approval="never",
)
async def preview_price_change(args, ctx):
    return ctx.platform.preview_price_change(
        args["product_id"],
        args["new_price"],
    )

@define_tool(
    "preview_availability_change",
    {"product_id": str, "available": bool},
    approval="never",
)
async def preview_availability_change(args, ctx):
    return ctx.platform.preview_availability_change(
        args["product_id"],
        args["available"],
    )
```

The dispatcher uses the same platform client, but it is not model-callable. It
runs only after the plan approval gate resolves to `"approve"`. Unlike tool
handlers, the dispatcher does not receive `services` through `ctx`; create it in
the same runtime factory and close over the API client:

```python
def dispatch_actions(actions, ctx):
    created_actions = []

    for action in actions:
        payload = action["payload"]

        if action["kind"] == "price_change":
            created_actions.append(
                api.create_action(
                    action_type="PRICE_CHANGE",
                    payload={
                        "productId": payload["product_id"],
                        "newPrice": payload["new_price"],
                    },
                )
            )
        elif action["kind"] == "availability_change":
            created_actions.append(
                api.create_action(
                    action_type="AVAILABILITY_CHANGE",
                    payload={
                        "productId": payload["product_id"],
                        "available": payload["available"],
                    },
                )
            )
        else:
            raise ValueError(f"unsupported action kind: {action['kind']}")

    return {
        "entitiesAffected": len(created_actions),
        "summary": f"Created {len(created_actions)} platform action(s)",
        "details": {
            "createdActions": created_actions,
        },
    }
```

Then wire both pieces into the runtime:

```python
runtime = create_runtime(
    define_runtime(
        tenant=define_tenant("acme", "v1"),
        agents=[coordinator, planner, executor],
        providers={"anthropic": {"apiKey": "unused"}},
    ),
    services={"platform": api},
    action_dispatcher=dispatch_actions,
)
```

In this setup, the model can call preview tools while planning, but it cannot
directly call `api.create_action(...)`. It can only store a typed plan and ask
the executor to run `executePlan`. The runtime validates the plan, asks for
approval when required, hydrates references, and only then invokes
`dispatch_actions`.

### Dependency flow: tools vs dispatchers

Tool handlers and action dispatchers both receive a `ctx` argument, but it is
not the same context.

Tool handler context is host-service context. If you pass a service through
`create_runtime(..., services={"platform": api})`, custom Python tools can read
it through `ctx.platform`, `ctx["platform"]`, or `ctx.services["platform"]`:

```python
@define_tool("preview_price_change", {"product_id": str, "new_price": float})
async def preview_price_change(args, ctx):
    # This works for tools.
    return ctx.platform.preview_price_change(
        args["product_id"],
        args["new_price"],
    )
```

Action dispatcher context is plan-execution context. It carries runtime-built
execution data, such as hydrated references, not the `services` mapping:

```python
def dispatch_actions(actions, ctx):
    # This works for dispatchers.
    resolved_refs = ctx["resolved_refs"]
```

So do not write dispatchers that expect service injection through `ctx`:

```python
def dispatch_actions(actions, ctx):
    # Wrong today: action dispatcher ctx is not the tool service context.
    return ctx.platform.create_action(...)
```

Instead, build the runtime in a factory, create the platform client once, and
close over it from the dispatcher:

```python
from flowai_harness import (
    create_runtime,
    define_runtime,
    define_specialist,
    define_tenant,
    define_tool,
)

def build_runtime(tenant_id: str):
    api = MockCommerceApi()  # in production: a per-tenant API client

    @define_tool("preview_price_change", {"product_id": str, "new_price": float})
    async def preview_price_change(args, ctx):
        return ctx.platform.preview_price_change(
            args["product_id"],
            args["new_price"],
        )

    def dispatch_actions(actions, ctx):
        created = []
        for action in actions:
            if action["kind"] == "price_change":
                created.append(
                    api.create_action(
                        action_type="PRICE_CHANGE",
                        payload=action["payload"],
                    )
                )

        return {
            "entitiesAffected": len(created),
            "summary": f"Created {len(created)} platform action(s)",
            "details": {"createdActions": created},
        }

    planner = define_specialist(
        "commercial_planner",
        model="claude-sonnet-4-6",
        prompt="Preview price changes before proposing a plan.",
        tools=[preview_price_change],
    )

    return create_runtime(
        define_runtime(
            tenant=define_tenant(tenant_id, "v1"),
            agents=[planner],
            providers={"anthropic": {"apiKey": "unused"}},
        ),
        services={"platform": api},
        action_dispatcher=dispatch_actions,
    )
```

Both callbacks use the same platform client, but the dependency path is
different:

```text
tool callback       -> ctx.platform
action dispatcher  -> closure / runtime factory scope
```

## Tool approvals

A tool can opt in through `define_tool(..., approval="always")` or a dynamic
predicate. The approval event shape is identical except `kind` is `"tool"` and
`target` carries the tool name. Approving runs the Python handler; rejecting
records a decision event and returns a tool result that signals the rejection
to the LLM.

```python
@define_tool("increment_counter", {"amount": int}, approval="always")
async def increment_counter(args, ctx):
    return {"count": args["amount"]}
```

For dynamic policies, the predicate receives `(args, ctx)` and returns a bool.
When it returns `True` the runtime gates the call; when it returns `False` the
call runs immediately.

```python
def needs_approval(args, ctx):
    return ctx["target"] == "guarded_echo"

@define_tool("guarded_echo", {"value": str}, approval=needs_approval)
async def guarded_echo(args, ctx):
    return {"echo": args["value"]}
```

## Common errors

| Error | Fix |
| --- | --- |
| No `approval-required` event appears | Check the resolved runtime, agent, and tool policy. `"never"` at a narrower scope can override a broader floor. |
| A plan `revise` response acts like a rejection | `revise` is plan-only. Tool approvals treat unsupported revise requests as rejection. |
| The stream appears paused | Respond with `runtime.respond_to_approval(...)`; gated work does not continue until the host sends a decision. |
| The wrong tool is gated | Tool overrides are scoped by agent, so check the agent that owns the current tool call. |

## See also

- [Agents concept](/docs/concepts/agents)
- [`Runtime` reference](/docs/reference/runtime#flowai_harness.runtime.Runtime)
- [`define_coordinator` reference](/docs/reference/agents#flowai_harness.agents.define_coordinator)
- [Streaming Events](/docs/guides/streaming)