Prefect

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

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 == "RateLimitGuard"
    return False

@chio_task(
    tool_server="web",
    retries=3,
    retry_delay_seconds=30,
    retry_condition_fn=retry_on_rate_limit,
)
def search_prices(sku: str) -> list[dict]: ...

from chio_adapter_base.redact import RedactionPolicy

FILESYSTEM_SCOPE = ChioScope(grants=[
    ToolGrant(server_id="workspace", tool_name="chio_file_write",
              operations=[Operation.INVOKE]),
])

# 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="workspace",
    tool_name="chio_file_write",
    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=FILESYSTEM_SCOPE,
    capability_id=os.environ["CHIO_FILESYSTEM_CAPABILITY_ID"],
    chio_client=authority,
    tool_server="workspace",
    redaction_policy=RedactionPolicy.chio_default(),
)
def fs_pipeline(path: str, content: bytes) -> None:
    write_file(path, content)

@chio_task(tool_server="web")
async def search_prices(sku: str) -> list[dict]:
    return await external_search.run(sku)

@chio_task(tool_server="web")
def summarize_prices(listings: list[dict]) -> dict:
    return analyzer.run(listings)

import pytest

from chio_sdk import ChioScope, Operation, ToolGrant
from chio.testing import allow_all, deny_all
from chio_prefect import chio_task, chio_flow

TEST_SCOPE = ChioScope(grants=[
    ToolGrant(server_id="pricing", tool_name="quote", operations=[Operation.INVOKE]),
])

PRICES = {"sku-118": "24.00", "sku-204": "58.00", "sku-301": "140.00"}

def test_pipeline_happy_path():
    client = allow_all()

    @chio_task(tool_server="pricing", chio_client=client)
    def quote(sku: str) -> str:
        return PRICES[sku]

    @chio_flow(
        scope=TEST_SCOPE,
        capability_id="cap-test",
        tool_server="pricing",
        chio_client=client,
    )
    def pipeline() -> str:
        return quote("sku-204")

    assert pipeline() == "58.00"

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="pricing", capability_id="cap-test", chio_client=client)
    def quote(sku: str) -> str:
        return PRICES[sku]

    with pytest.raises(PermissionError):
        quote("sku-204")

sdks/python/chio-prefect/
  pyproject.toml          # deps: the Chio Python SDK, 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