Temporal

Why Temporal Through Chio

Chio evaluates a capability against a policy and produces a signed receipt. Temporal adds capabilities that a single evaluation does not: persistence, retry with backoff, saga compensation, visibility, and multi-worker distribution. Each Temporal activity that performs a tool call should pass through the Chio kernel for capability validation and receipt signing. A workflow-level grant bounds the workflow, and per-activity evaluation enforces that bound.

Install

Install the package from its own directory in a clone, together with the chio-sdk-python distribution beside it: that is the one that provides the chio_sdk module every snippet on this page imports. The first three commands build the environment the package's own test suite runs in; the last three are that suite.

uv venv --python 3.11
uv pip install -e '.[dev]'
uv pip install -e ../chio-sdk-python

uv run pytest
uv run mypy src/
uv run ruff check src/ tests/

The declared runtime dependencies are chio-sdk-python>=0.1.0, chio-adapter-base>=0.2.0,<0.3, temporalio>=1.7,<2, and pydantic>=2.5,<3. The first two are resolved from sibling directories: the pyproject pins them under [tool.uv.sources] as ../chio-sdk-python and ../chio-adapter-base. The chio sidecar runs next to your Temporal worker.

Public API

Deployment Topology

The Chio sidecar runs alongside the activity worker: same pod in Kubernetes, same host in VM deployments. Workflow workers do not call the sidecar directly; they orchestrate. Activity workers are the enforcement point, because Temporal's execution model treats activities as the side-effect boundary.

Activity Interceptor

The primary path is the activity interceptor. Register one on the worker; every activity execution is automatically evaluated through the chio sidecar using the capability from the registered WorkflowGrant.

async def evaluate_tool_call(
    self,
    *,
    capability_id: str,
    tool_server: str,
    tool_name: str,
    parameters: dict[str, Any],
) -> NoReturn:

from chio_sdk import ChioClient
from chio_sdk.models import CapabilityToken, ChioScope, Operation, ToolGrant
from chio_temporal import ChioActivityInterceptor, WorkflowGrant
from temporalio.client import Client
from temporalio.worker import Worker


def scope_for(*tools: str, server_id: str = "agent-tools") -> ChioScope:
    """Local helper. The SDK ships no scope_for; it ships ChioScope."""
    return ChioScope(
        grants=[
            ToolGrant(server_id=server_id, tool_name=tool, operations=[Operation.INVOKE])
            for tool in tools
        ]
    )


chio = ChioClient("http://127.0.0.1:9090")
client = await Client.connect("localhost:7233")

interceptor = ChioActivityInterceptor(
    chio_client=chio,
    sidecar_url="http://127.0.0.1:9090",
    # Fallback tool_server id when an activity has no explicit mapping.
    default_tool_server="agent-tools",
    # Per-activity-type override. Activities normally map 1:1 to a
    # chio tool server in production deployments.
    activity_tool_server_map={
        "call_tool":     "agent-tools",
        "read_database": "db-readonly",
        "send_email":    "email-outbound",
    },
    # Optional: drain finalised workflow receipts to a sink.
    receipt_sink=my_receipt_sink,
    # Optional: override argument redaction. Defaults to
    # RedactionPolicy.chio_default() when omitted.
    redaction_policy=None,
)

# Before starting the workflow, register the grant for its workflow_id.
token: CapabilityToken = await chio.create_capability(
    subject="agent:my-pipeline",
    scope=scope_for("call_tool", "read_database", "send_email"),
    ttl_seconds=3600,
)
grant = WorkflowGrant(
    workflow_id="agent-run-123",
    token=token,
    tool_server="agent-tools",
)
interceptor.register_workflow_grant(grant)

worker = Worker(
    client,
    task_queue="agent-tasks",
    workflows=[AgentWorkflow],
    activities=[call_tool, read_database, send_email],
    interceptors=[interceptor],
)

The interceptor looks up the grant when the first activity for a given (workflow_id, run_id) pair executes. It then evaluates each activity against the sidecar using the grant's capability id and records a WorkflowStepReceipt on an in-flight WorkflowReceipt. Concurrent workflows on the same worker do not stomp each other's step lists; state is keyed on (workflow_id, run_id).

Argument Redaction

Activity parameters are redacted in the activity layer before the sidecar ever sees them, so workflow determinism is unaffected. The interceptor binds Temporal's activity wire shape (arguments arrive as parameters["args"]) through chio_adapter_base.redact.bind_and_redact and the DEFAULT_TOOL_POSITIONAL_NAMES registry, so positional bodies for tools like chio_file_write and chio_file_edit are bound to their parameter names and stubbed. The default policy replaces protected fields with {"omitted": true, "byte_count": N}; the activity body itself still runs with the original arguments. Pass a redaction_policy to the constructor to override what is stubbed.

Convenience Builder

build_chio_worker wires the standard pieces in one call: mint a capability (or reuse a supplied one), construct the interceptor, register the grant, build the worker, and return the trio.

from chio_temporal import build_chio_worker

worker, interceptor, grant = await build_chio_worker(
    client,
    task_queue="agent-tasks",
    activities=[call_tool, read_database, send_email],
    workflows=[AgentWorkflow],
    chio_client=chio,
    workflow_id="agent-run-123",
    # capability_id is keyword-only with no default, so it is always passed.
    # Empty here because scope wins: the builder mints a fresh token and keeps
    # this value only as grant.metadata["supplied_capability_id"].
    capability_id="",
    scope=scope_for("call_tool", "read_database", "send_email"),
    subject="agent:my-pipeline",
    ttl_seconds=3600,
    default_tool_server="agent-tools",
    activity_tool_server_map={"send_email": "email-outbound"},
)

await worker.run()

The two ways to supply authority are not symmetrical, and the asymmetry is easy to trip over. capability_id is keyword-only with no default (chio_temporal/worker.py:40), so leaving it out is a TypeError rather than a fallback; the builder only refuses when both it and scope are absent (worker.py:120-123). And when both are present, scope wins: the builder mints a new token and demotes the id you passed to grant_metadata["supplied_capability_id"] (worker.py:125-141), so a worker configured with both does not run under the capability you named. To reuse a pre-minted id, pass it and leave scope and subject out; the builder then wraps the id in a stand-in token whose only meaningful field is the id the interceptor evaluates against (worker.py:133-139).

Workflow Grants and Attenuation

A WorkflowGrant pins a capability token to a Temporal workflow_id. Optional run_id pinning lets a grant bind to one execution; otherwise it applies across every run of the workflow id. attenuate_for_activity is the intended way to hand one activity a narrower grant:

narrower = await grant.attenuate_for_activity(
    chio,
    new_scope=scope_for("read_database"),
    tool_server="db-readonly",
)

# Register a per-activity-type hook that returns the narrower grant.
interceptor.register_activity_grant_override(
    "read_database",
    lambda info: narrower,
)

The hook is invoked with the Temporal activity.Info for each execution; returning None falls back to the workflow-level grant. The interceptor verifies the override scope is a subset of the workflow grant scope and raises ChioTemporalConfigError if not.

Deny Handling

When the sidecar returns a deny verdict, the interceptor records the deny receipt on the workflow's WorkflowReceipt and raises a non-retryable ApplicationError with type "ChioCapabilityDenied". Sagas can catch the error and run compensations:

from temporalio.exceptions import ApplicationError
from chio_temporal import DENIED_ERROR_TYPE

@workflow.defn
class TransferWorkflow:
    @workflow.run
    async def run(self, transfer: Transfer) -> TransferResult:
        compensations: list[tuple[str, list]] = []
        debit = await workflow.execute_activity(
            debit_account,
            args=[transfer.source, transfer.amount],
            start_to_close_timeout=timedelta(seconds=30),
        )
        compensations.append(("credit_account", [transfer.source, transfer.amount]))

        try:
            credit = await workflow.execute_activity(
                credit_account,
                args=[transfer.destination, transfer.amount],
                start_to_close_timeout=timedelta(seconds=30),
            )
        except ApplicationError as exc:
            if exc.type == DENIED_ERROR_TYPE:
                for name, args in reversed(compensations):
                    await workflow.execute_activity(
                        name, args=args,
                        start_to_close_timeout=timedelta(seconds=30),
                    )
            raise

        return TransferResult(debit=debit, credit=credit)

Workflow Receipt Aggregation

The interceptor maintains a WorkflowReceipt per (workflow_id, run_id) pair. Each executed activity appends a WorkflowStepReceipt. Call finalize_workflow when the workflow completes, then flush_workflow_receipt to forward the envelope to the configured sink:

# At workflow completion (typically from a "finalise" activity).
interceptor.finalize_workflow(
    workflow_id="agent-run-123",
    run_id=run_id,
    outcome="success",   # or "failure" | "cancelled"
)
envelope = await interceptor.flush_workflow_receipt(
    workflow_id="agent-run-123",
    run_id=run_id,
)
# envelope["version"] == "chio-temporal/v1"
# envelope has: workflow_id, run_id, parent_workflow_ids, started_at,
# completed_at, outcome, step_count, allow_count, deny_count, steps, metadata

The activity result is not touched. The interceptor returns whatever the next link in the chain returns (chio_temporal/interceptor.py:321), so nothing is appended to it and no receipt id rides back in the value your activity produced. The receipt id lands on the in-memory WorkflowStepReceipt instead, via _record_step, and the receipt body stays in the Chio receipt store.

On the allow path that keeps workflow history free of receipt content. On the deny path it does not: the ApplicationError the interceptor raises carries ChioTemporalError.to_dict() as its details (interceptor.py:494-499), and that dict includes the receipt id and the decision (chio_temporal/errors.py:42, 54, 59-60), which Temporal persists in history like any other failure detail.

Where the Integration Stops

The interceptor governs the synchronous allow and deny path. Four things sit outside it:

• Approval inside an activity. The interceptor does not hold an Activity open on a pending_approval verdict. Activity code that wants a verdict in hand calls the client directly and decides for itself whether to await a Signal. Reach for evaluate_tool_call_advisory for a non-authoritative observation, or evaluate_tool_call_mediated when you hold a signed token and want an authoritative one (chio_sdk/client.py:549, 591). Not evaluate_tool_call: that one is the id-only wrapper and always raises.
• Budget reversal. A budget charge is not unwound when a workflow fails. Reverse it with a compensating activity of your own, the same way you compensate any other side effect.
• Non-Python workers. The interceptor is Python. A temporal-sdk-core worker in another language reaches the same policy through the sidecar's language-neutral HTTP surface.
• Multi-cluster replication. Temporal replicates across clusters; a Chio receipt log is per-kernel. Receipt continuity across a Temporal cluster failover is an open design question upstream (docs/protocols/TEMPORAL-INTEGRATION.md:317-319), with no code or decision behind it yet.

Package Layout

sdks/python/chio-temporal/
  pyproject.toml                # deps: chio-sdk-python, chio-adapter-base, temporalio
  src/chio_temporal/
    __init__.py                 # public API (see table above)
    interceptor.py              # ChioActivityInterceptor, ActivityGrantOverride, DENIED_ERROR_TYPE
    grants.py                   # WorkflowGrant (+ attenuate_for_activity)
    receipt.py                  # WorkflowReceipt, WorkflowStepReceipt, ENVELOPE_VERSION
    worker.py                   # build_chio_worker
    errors.py                   # ChioTemporalError, ChioTemporalConfigError
    py.typed
  tests/
    test_interceptor.py
    test_workflow_receipt.py
    test_redaction.py
    test_temporal_redaction_helpers.py

Next Steps

• Kafka · enforce capabilities on event-driven producers and consumers
• Budgets · how capability scope and per-invocation cost caps are set at mint time
• LangGraph · pair durable workflows with graph-based agent orchestration