Chio/Docs

BuildOrchestrators

Prefect

Use chio-prefect decorators to evaluate tasks before they run, emit Prefect decision events, and limit enclosed tasks with flow scope.

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 alonePrefect + Chio
Tasks retry on failureTasks are denied before side-effects when a capability is revoked
Flow parameters gate inputsFlow scope gates the set of tools every enclosed task may call
Events record task state transitionsEvents record signed receipts linked to flow-run and task-run ids
Access control is deployment-levelAccess 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 pulls in chio-sdk-python, the distribution that provides the chio_sdk module the decorators import. The chio sidecar runs next to your Prefect worker; no new processes to manage if you already have a sidecar for other SDKs.

sdks/python/chio-prefect/README.md:13-15bash
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. There is no shared client between task calls, though: unless you pass your own chio_client=, each decorated invocation constructs a ChioClient and closes it in a finally (sdks/python/chio-prefect/src/chio_prefect/decorators.py:90-93, 403, 417), so inject one when a worker runs many concurrent tasks.

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:

EventWhen emittedPayload
chio.receipt.allowBefore task body runs, on allowReceipt id, capability id, tool server, tool name, timestamp
chio.receipt.denyBefore task body runs, on denyReceipt id, guard name, deny reason, capability id, tool server, tool name

The event's subject is the receipt, not the task run. The Prefect task and flow ids ride in the related list beside it, which is what puts the event on the right row in the Prefect UI:

sdks/python/chio-prefect/src/chio_prefect/events.py85-124python
def _receipt_resource(receipt: ChioReceipt) -> dict[str, str]:
    """Build the :class:`prefect.events.Resource` dict for a receipt."""
    return {
        "prefect.resource.id": f"chio.receipt.{receipt.id}",
        "prefect.resource.role": "chio-receipt",
        "chio.capability_id": receipt.capability_id or "",
        "chio.tool_server": receipt.tool_server or "",
        "chio.tool_name": receipt.tool_name or "",
    }


def _task_related(
    *,
    task_name: str,
    flow_run_id: str | None,
    task_run_id: str | None,
) -> list[dict[str, str]]:
    """Build the related-resource list tying the event to the Prefect task run.

    The ids are used by Prefect's UI to draw the event on the correct
    flow-run / task-run timeline. Empty strings are omitted so we do not
    emit malformed related-resource entries.
    """
    related: list[dict[str, str]] = []
    if flow_run_id:
        related.append(
            {
                "prefect.resource.id": f"prefect.flow-run.{flow_run_id}",
                "prefect.resource.role": "flow-run",
            }
        )
    if task_run_id:
        related.append(
            {
                "prefect.resource.id": f"prefect.task-run.{task_run_id}",
                "prefect.resource.role": "task-run",
                "prefect.resource.name": task_name,
            }
        )
    return related

So one allow event goes out shaped like this, with resource, payload and related all passed to prefect.events.emit_event (events.py:59-64, 141-163):

python
# A receipt emitted on the Prefect events backend
{
    "event": "chio.receipt.allow",
    "resource": {
        "prefect.resource.id": "chio.receipt.9f2c0a1d...c41e",
        "prefect.resource.role": "chio-receipt",
        "chio.capability_id": "cap-research-pipeline",
        "chio.tool_server": "search-srv",
        "chio.tool_name": "search_documents",
    },
    "payload": {
        "receipt_id": "9f2c0a1d...c41e",
        "verdict": "allow",
        "capability_id": "cap-research-pipeline",
        "tool_server": "search-srv",
        "tool_name": "search_documents",
        "task_name": "search_documents",
        "timestamp": 1776631242,
    },
    "related": [
        {"prefect.resource.id": "prefect.flow-run.<flow-run-id>",
         "prefect.resource.role": "flow-run"},
        {"prefect.resource.id": "prefect.task-run.<task-run-id>",
         "prefect.resource.role": "task-run",
         "prefect.resource.name": "search_documents"},
    ],
}

Two shapes there are easy to get wrong. receipt_id is a Chio receipt id, which is always 64 lowercase hex characters (chio_sdk/_generated/receipt/record_schema.py:216-218 enforces the pattern); it is elided in the middle above and is never a ULID. And timestamp is copied straight off the receipt, where it is Unix seconds as an integer (record_schema.py:219-221), not an ISO-8601 string.

Because the flow-run and task-run ids arrive as related resources, the Prefect UI renders the event on the correct row and you can pivot from a receipt back to the exact flow run. If the events backend is unavailable the decorator logs at INFO (events.py:74-82); 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 (asyncio.run, decorators.py:334-337), which blocks the calling worker thread until the verdict is in. Inside that loop the sync body itself is handed to a thread (await asyncio.to_thread(fn, *args, **kwargs), decorators.py:421-422), so a long-running task body never blocks the loop.

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")

    # capability_id is required on a task called outside a flow: without one
    # the decorator raises ChioPrefectConfigError before the sidecar is asked.
    @chio_task(tool_server="srv", capability_id="cap-test", 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-python, chio-adapter-base, prefect >= 3, pydantic
  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
    py.typed
  tests/
    test_task_decorator.py    # decorators, deny translation, event emission
    test_flow_attenuation.py  # flow scope and the task-scope subset check
    test_redaction.py         # the redaction policy, by some distance the largest

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
PreviousTemporalNextKafka