Prefect · Chio Docs

Why Prefect Through Chio

Prefect provides features common in agent pipelines: typed function decorators, retry policies, async-first execution, a UI that shows where time is going. It does not provide a policy mechanism. Scheduling a flow does not restrict its tool calls, and @task does not model authority. The chio integration adds policy evaluation.

Prefect alone	Prefect + Chio
Tasks retry on failure	Tasks are denied before side-effects when a capability is revoked
Flow parameters gate inputs	Flow scope gates the set of tools every enclosed task may call
Events record task state transitions	Events record signed receipts linked to flow-run and task-run ids
Access control is deployment-level	Access control is per-tool, per-scope, time-bounded, revocable
Run history answers "what happened"	Receipt chain answers "what was allowed and why"

Install

The package targets Prefect 3 and ships alongside chio-sdk. The chio sidecar runs next to your Prefect worker; no new processes to manage if you already have a sidecar for other SDKs.

bash

uv pip install chio-prefect
# or
pip install chio-prefect

No environment variables are required. The default sidecar URL is http://127.0.0.1:9090; pass sidecar_url= on any decorator to override.

Deployment Topology

Prefect workers run the task bodies, so they are the enforcement point. The Chio kernel runs as a sidecar on the same host or pod; the decorators speak HTTP to it. Prefect's API server stays untouched, its events backend stays untouched, and the Chio receipt log stays external to flow-run history.

rendering…

Workers call the Chio sidecar before every decorated task body runs. Decisions flow back as Prefect events on the same flow-run timeline.

One sidecar per worker, not per flow

The sidecar is a per-host enforcement daemon, not a per-flow attachment. A Prefect worker running many concurrent flows shares a single sidecar; concurrency is bounded by the sidecar's HTTP pool, not by the decorator.

Quickstart

Two decorators: @chio_task wraps a Prefect task, @chio_flow wraps a Prefect flow. Task bodies receive an allow/deny verdict before they run, and the verdict is emitted as a Prefect event tied to the task-run id.

python

from chio_sdk.client import ChioClient
from chio_sdk.models import ChioScope, Operation, ToolGrant
from chio_prefect import chio_flow, chio_task

PIPELINE_SCOPE = ChioScope(
    grants=[
        ToolGrant(
            server_id="search-srv",
            tool_name="search_documents",
            operations=[Operation.INVOKE],
        ),
        ToolGrant(
            server_id="search-srv",
            tool_name="analyze_results",
            operations=[Operation.INVOKE],
        ),
    ]
)

@chio_task(tool_server="search-srv")
def search_documents(query: str) -> list[dict]:
    return external_search.run(query)

@chio_task(tool_server="search-srv")
def analyze_results(documents: list[dict]) -> dict:
    return analyzer.run(documents)

@chio_flow(
    scope=PIPELINE_SCOPE,
    capability_id="cap-research-pipeline",
    tool_server="search-srv",
)
def research_pipeline(query: str) -> dict:
    docs = search_documents(query)
    return analyze_results(docs)

research_pipeline("capability-based security")

The decorators are thin wrappers over Prefect's own: every option you would pass to @task or @flow passes through verbatim. Retries, timeouts, task runners, tags, custom result storage, all of it still works.

Flow Scope and Attenuation

The flow scope is a capability envelope. Tasks inside a flow cannot exceed it, which means a deployment that schedules research_pipeline with search and analysis grants cannot have one of its tasks silently acquire a file.write capability. The subset check is enforced at call time:

python

# Allowed: task scope is a subset of flow scope
@chio_task(
    scope=ChioScope(grants=[
        ToolGrant(server_id="search-srv", tool_name="search_documents",
                  operations=[Operation.INVOKE]),
    ]),
    tool_server="search-srv",
)
def search_documents(query: str) -> list[dict]: ...

# Denied at call time: task scope escapes the flow envelope
@chio_task(
    scope=ChioScope(grants=[
        ToolGrant(server_id="fs-srv", tool_name="write_file",
                  operations=[Operation.INVOKE]),
    ]),
    tool_server="fs-srv",
)
def write_file(path: str, body: bytes) -> None: ...

Tasks that omit scope inherit the enclosing flow's scope. Standalone tasks (tasks called outside any @chio_flow) must declare their own capability_id; omitting it raises ChioPrefectConfigError on the task's first call. Decoration always succeeds; the check runs when the wrapped task is invoked, but still fails client-side before any sidecar call, so it raises a configuration error instead of a kernel deny.

Use flow scope for authorization, task scope for documentation

The flow scope is the scope a deployment owner approves; task scope is how a developer documents which grant a task actually needs. Keeping task scope narrower than flow scope is what makes the subset check meaningful.

Receipts as Prefect Events

Every task evaluation produces a receipt, and every receipt is mirrored as a Prefect event on the task-run timeline. Two event names are used:

Event	When emitted	Payload
`chio.receipt.allow`	Before task body runs, on allow	Receipt id, capability id, tool server, tool name, timestamp
`chio.receipt.deny`	Before task body runs, on deny	Receipt id, guard name, deny reason, full decision dict

python

# A receipt emitted on the Prefect events backend
{
    "event": "chio.receipt.allow",
    "resource": {
        "prefect.resource.id": "prefect.task-run.<task-run-id>",
        "prefect.flow-run.id": "<flow-run-id>",
    },
    "payload": {
        "receipt_id": "01HXYZ...7K4",
        "verdict": "allow",
        "capability_id": "cap-research-pipeline",
        "tool_server": "search-srv",
        "tool_name": "search_documents",
        "task_name": "search_documents",
        "timestamp": "2026-04-19T20:14:02.118Z",
    },
}

Events are resource-linked to the task-run and flow-run ids, which means the Prefect UI renders them on the correct row and you can pivot from a receipt back to the exact flow run in one query. If the events backend is unavailable the decorator logs at INFO; it will not silently drop a receipt.

Argument Redaction

Before task parameters are sent to the sidecar for evaluation, the decorator runs them through chio_adapter_base.redact.bind_and_redact. Positional and keyword arguments are bound to their parameter names, and protected body fields, for example chio_file_write.content and chio_file_edit.patch, are replaced with {"omitted": true, "byte_count": N}. The task body still runs with the original, unredacted arguments; only what the sidecar evaluates, and therefore what appears in receipts and Prefect event payloads, is stubbed.

python

from chio_adapter_base.redact import RedactionPolicy

# Both decorators accept redaction_policy. Omit it for the chio default
# (which stubs chio_file_write.content / chio_file_edit.patch). A custom
# policy fully replaces the default.
@chio_task(
    tool_server="fs-srv",
    redaction_policy=RedactionPolicy.chio_default(),
)
def write_file(path: str, content: bytes) -> None:
    ...

# A @chio_flow's redaction_policy becomes the default for every enclosed
# task that does not set its own.
@chio_flow(
    scope=PIPELINE_SCOPE,
    capability_id="cap-fs-pipeline",
    tool_server="fs-srv",
    redaction_policy=RedactionPolicy.chio_default(),
)
def fs_pipeline(path: str, content: bytes) -> None:
    write_file(path, content)

Resolution order for the effective policy is per-task redaction_policy > the enclosing flow's policy > RedactionPolicy.chio_default().

Denials and Retries

A deny verdict raises PermissionError from the task's wrapper, which Prefect marks as a task failure. The decorator attaches the structured chio verdict directly to the exception as PermissionError.chio_error (a ChioPrefectError), so except PermissionError is the canonical catch and downstream code can distinguish a policy failure from a business failure:

python

from chio_prefect import ChioPrefectError

try:
    research_pipeline("sensitive query")
except PermissionError as e:
    err: ChioPrefectError | None = getattr(e, "chio_error", None)
    if err is not None:
        log.warning(
            "denied by chio",
            receipt_id=err.receipt_id,
            guard=err.guard,
            reason=err.reason,
        )
    else:
        raise

Read chio_error, not __cause__

Attach detection to PermissionError.chio_error, not __cause__. On a policy deny the wrapper raises the PermissionError with no from clause, so __cause__ is unset; on a sidecar 403 the cause is a ChioDeniedError, not a ChioPrefectError. Only chio_error is populated on both paths.

By default, denials are not retried: replaying a policy decision that already returned "deny" with the same inputs is pointless and spends evaluation budget. If your guard depends on state that can change between attempts, opt into retry with a retry_condition_fn on the task:

python

def retry_on_rate_limit(task, task_run, state) -> bool:
    exc = state.result(raise_on_failure=False)
    if isinstance(exc, PermissionError):
        err = getattr(exc, "chio_error", None)
        if err is not None:
            return err.guard == "rate-limit-guard"
    return False

@chio_task(
    tool_server="search-srv",
    retries=3,
    retry_delay_seconds=30,
    retry_condition_fn=retry_on_rate_limit,
)
def search_documents(query: str) -> list[dict]: ...

Async and Sync

Both shapes are supported and the decorator preserves the function's signature. An async def becomes an async Prefect task; a plain def becomes a sync task whose chio evaluation is driven from a throwaway event loop so the worker is never blocked.

python

@chio_task(tool_server="search-srv")
async def search_documents(query: str) -> list[dict]:
    return await external_search.run(query)

@chio_task(tool_server="search-srv")
def analyze_results(documents: list[dict]) -> dict:
    return analyzer.run(documents)

Testing

The chio_sdk.testing helpers include allow_all() and deny_all() mock clients. Inject them via the chio_client= parameter on the decorator so flows can be unit-tested without a live sidecar:

python

from chio_sdk.testing import allow_all, deny_all
from chio_prefect import chio_task, chio_flow

def test_pipeline_happy_path():
    client = allow_all()

    @chio_task(tool_server="srv", chio_client=client)
    def double(x: int) -> int:
        return x * 2

    @chio_flow(
        scope=PIPELINE_SCOPE,
        capability_id="cap-test",
        tool_server="srv",
        chio_client=client,
    )
    def pipeline() -> int:
        return double(21)

    assert pipeline() == 42

def test_pipeline_denied():
    client = deny_all(reason="budget exceeded")

    @chio_task(tool_server="srv", chio_client=client)
    def double(x: int) -> int:
        return x * 2

    with pytest.raises(PermissionError):
        double(21)

Package Layout

text

sdks/python/chio-prefect/
  pyproject.toml          # deps: chio-sdk, prefect >= 3
  src/chio_prefect/
    __init__.py           # chio_task, chio_flow, errors, events
    decorators.py         # task and flow wrappers; flow-scope ContextVar
    events.py             # Prefect event emission
    errors.py             # ChioPrefectError, ChioPrefectConfigError
  tests/
    test_task_decorator.py
    test_flow_attenuation.py
    test_event_emission.py

Next Steps

Temporal · the durable workflow counterpart, with workflow-level grants and saga compensation
LangGraph · graph-based agent orchestration with the same kernel API
Budgets · attach spend envelopes to a flow and reconcile them on deny or failure
Receipt format · the payload shape mirrored into Prefect events

PreviousTemporal NextKafka