Temporal Integration

What is Temporal?

Temporal is a platform for durable execution: you write Workflows, deterministic orchestration code whose progress is replayed from event history and survives process restarts, retries, and long waits, and Activities, the non-deterministic work such as tool calls, HTTP requests, or model inference. The service assigns tasks from task queues to Workers, persists workflow state, and gives you reliable, observable automation for multi-step AI agents without hand-rolling sagas or bespoke recovery logic.

For a guided introduction, see Understanding Temporal.

How Tenuo fits

Access control in Temporal typically relies on worker identity or task queue tokens, which grant the same permissions to every workflow running on that worker. As agents take on more consequential actions, you need finer control: which tools each agent may use for a given task, with what arguments, and on whose authority.

Tenuo adds that task-scoped authorization layer. A signed warrant travels with each workflow and is verified by the worker interceptor before every Activity runs. Agents are constrained by design: an agent can only execute what its warrant permits, and every action is cryptographically attributable to the entity that authorized it. Activity definitions require no changes, and verification runs in-process with no external service dependency.

Prerequisites

• Familiarity with Temporal Workflows, Activities, and Workers (What is Temporal above, or Temporal learning resources).
• A running Temporal cluster (local temporal server start-dev or Temporal Cloud).
• Python 3.10+ (inherited from temporalio>=1.23.0, which provides SimplePlugin).

Install

uv pip install "tenuo[temporal]"

This installs temporalio>=1.23.0 and tenuo_core, a compiled Rust extension with prebuilt wheels for common platforms.

Configure Workers to use Tenuo

Add the TenuoTemporalPlugin to your Client. The plugin wires client interceptors, worker interceptors, and the workflow sandbox runner in one step.

from temporalio.client import Client
from temporalio.worker import Worker
from tenuo import SigningKey
from tenuo.temporal import TenuoTemporalPlugin, TenuoPluginConfig, EnvKeyResolver

# For local development — generate a key pair:
control_key = SigningKey.generate()
issuer_public_key = control_key.public_key

plugin = TenuoTemporalPlugin(
    TenuoPluginConfig(
        key_resolver=EnvKeyResolver(),
        trusted_roots=[issuer_public_key],
    )
)

client = await Client.connect("localhost:7233", plugins=[plugin])
worker = Worker(
    client,
    task_queue="my-queue",
    workflows=[MyWorkflow],
    activities=[read_file, write_file],
)

TenuoPluginConfig requires two things:

• trusted_roots — public keys of warrant issuers (for verification on the activity worker)
• key_resolver — how the workflow worker fetches holder signing keys for PoP (e.g. EnvKeyResolver, VaultKeyResolver, AWSSecretsManagerKeyResolver)

EnvKeyResolver maps key_id to environment variables using the convention TENUO_KEY_<key_id> with base64-encoded signing key bytes:

# Generate and export a key for local development:
export TENUO_KEY_agent1=$(python -c "from tenuo import SigningKey; import base64; k=SigningKey.generate(); print(base64.b64encode(k.secret_key_bytes()).decode())")

TenuoTemporalPlugin automatically preloads all TENUO_KEY_* variables into an in-memory cache so that key resolution never touches os.environ inside the workflow sandbox. For production, use VaultKeyResolver or AWSSecretsManagerKeyResolver instead. Tenuo Cloud handles key issuance, warrant minting, rotation, and audit for teams that prefer a managed control plane. See the reference for key management details.

Important: Pass the plugin on Client.connect(plugins=[plugin]) only. Workers created from that client automatically merge client plugins — do not duplicate.

About the names. Two public classes have similar names on purpose — they are not the same:

Class	Type	When to use
tenuo.temporal_plugin.TenuoTemporalPlugin	Temporal SDK SimplePlugin	Default. Pass to Client.connect(plugins=[...]). Wires the client interceptor, worker interceptor, and sandboxed workflow runner in one step.
tenuo.temporal.TenuoWorkerInterceptor	Temporal SDK WorkerInterceptor	Advanced only. Use when you are hand-composing your own Plugin / SimplePlugin and just want Tenuo’s authorization interceptor.

Define activities and workflows

Activity definitions stay unchanged — no Tenuo imports needed:

from pathlib import Path
from temporalio import activity, workflow
from datetime import timedelta

@activity.defn
async def read_file(path: str) -> str:
    return Path(path).read_text()

@activity.defn
async def write_file(path: str, content: str) -> str:
    Path(path).write_text(content)
    return f"Wrote {len(content)} bytes"

Use AuthorizedWorkflow to fail fast if warrant headers are missing, or use plain workflow.execute_activity() — the interceptor handles authorization either way:

from tenuo.temporal import AuthorizedWorkflow

@workflow.defn
class MyWorkflow(AuthorizedWorkflow):
    @workflow.run
    async def run(self, input_path: str) -> str:
        data = await self.execute_authorized_activity(
            read_file,
            args=[input_path],
            start_to_close_timeout=timedelta(seconds=30),
        )
        return data.upper()

Start an authorized workflow

Pass a warrant and key ID when starting a workflow. The warrant defines what the agent is allowed to do — your control plane or policy layer mints it.

from tenuo.temporal import execute_workflow_authorized

result = await execute_workflow_authorized(
    client=client,
    workflow_run_fn=MyWorkflow.run,
    workflow_id="process-001",
    warrant=warrant,
    key_id="agent1",
    args=["/data/input/report.txt"],
    task_queue="my-queue",
)

For long-running workflows where you need a handle to signal or query later, use start_workflow_authorized():

from tenuo.temporal import start_workflow_authorized

handle = await start_workflow_authorized(
    client=client,
    workflow_run_fn=ApprovalWorkflow.run,
    workflow_id="approval-001",
    warrant=warrant,
    key_id="agent1",
    args=[request_data],
    task_queue="my-queue",
)

# Signal, query, or await later
await handle.signal(ApprovalWorkflow.approve, decision)
result = await handle.result()

Capability constraints

Warrants use a closed-world (zero-trust) model: every argument the activity receives must be declared in the capability, even if unconstrained. Use Wildcard() for arguments that can take any value:

from tenuo import Warrant, SigningKey, Subpath, UrlSafe, Wildcard

warrant = (
    Warrant.mint_builder()
    .holder(agent_key.public_key)
    .capability("read_file", path=Subpath("/data"))       # path must be under /data
    .capability("search", query=Wildcard())               # query can be anything
    .capability("fetch_url",
        url=UrlSafe(allow_schemes=["https"],
                    allow_domains=["api.example.com"],
                    block_private=True),
        timeout=Wildcard(),                               # unconstrained but declared
    )
    .ttl(3600)
    .mint(control_key)
)

If an activity argument is not listed in the capability, the interceptor rejects the call with TemporalConstraintViolation — even if the value would otherwise be valid. This prevents accidental exposure of undeclared parameters.

Common mistake: listing only the constrained fields. If your activity has parameters path and encoding, the capability needs both — e.g. .capability("read_file", path=Subpath("/data"), encoding=Wildcard()).

See Temporal Integration Reference — Constraint Types for the full list of constraint types (Subpath, UrlSafe, Exact, Pattern, Range, AnyOf, etc.).

How it works

sequenceDiagram
    participant C as Client
    participant T as Temporal
    participant WW as Workflow Worker
    participant KR as KeyResolver
    participant AW as Activity Worker

    C->>T: execute_workflow(headers: warrant + key_id)
    T->>WW: workflow task
    WW->>KR: resolve(key_id)
    KR-->>WW: signing_key — never transmitted
    Note over WW: PoP = sign(warrant_id, tool, sorted_args, window_ts)
    WW->>AW: activity headers (warrant + PoP)
    Note over AW: verify warrant chain → trusted_roots
    Note over AW: verify PoP signature + constraints
    AW->>AW: execute activity (authorized)

The signing key is resolved on the worker and never leaves it. PoP is computed at schedule time (binding exact tool and args), then verified on the activity worker before execution. This works in both single-process demos and distributed deployments.

Child workflow delegation

Attenuate warrants when spawning child workflows so children get least-privilege access:

from tenuo.temporal import tenuo_execute_child_workflow

@workflow.defn
class ParentWorkflow:
    @workflow.run
    async def run(self) -> str:
        # Parent has read_file + write_file.
        # Child gets only read_file with a shorter TTL.
        return await tenuo_execute_child_workflow(
            ChildWorkflow.run,
            tools=["read_file"],
            ttl_seconds=60,
            args=["/data/input"],
            id=f"child-{workflow.info().workflow_id}",
            task_queue=workflow.info().task_queue,
        )

Important: workflow.execute_child_workflow() does not propagate warrant headers. Always use tenuo_execute_child_workflow() for authorized children.

Security

Fail-closed by default. Missing or invalid warrants block execution. Each activity dispatch includes a Proof-of-Possession (PoP) signature binding the tool name and arguments to the holder key. Enforcement is in-process (no Tenuo network hop at verify time).

Private keys never leave your infrastructure. Only the key_id and warrant material travel in Temporal headers. Workers resolve signing keys from your Vault, AWS Secrets Manager, GCP Secret Manager, or (for development) environment variables via KeyResolver. No private key material is transmitted to the Temporal cluster or any Tenuo endpoint.

Warrant chain verification. When warrants are attenuated (e.g. for child workflows), the full delegation chain is validated back to trusted roots, ensuring no intermediate warrant was forged or widened.

Authorization failures are wrapped in Temporal’s ApplicationError(non_retryable=True) to prevent retrying permanent denials.

Trust boundaries:

For the full threat model, PoP time windows, replay protection, root rotation, and revocation, see Temporal Integration Reference.

Activity summaries in the Temporal Web UI

The plugin enriches every authorized activity with a human-readable summary in the Temporal Web UI’s Event History:

from tenuo.temporal import tenuo_execute_activity

await tenuo_execute_activity(
    read_file,
    args=["/data/report.txt"],
    start_to_close_timeout=timedelta(seconds=30),
    summary="monthly sales report",
)

Runnable examples

These scripts under tenuo-python/examples/temporal/ are the fastest path to a working demo. Run temporal server start-dev in one terminal, then run the Python file in another.

Next steps

• Temporal Integration Reference — production checklist, key management (Vault, AWS, GCP), sandbox details, PoP mechanics, configuration reference, constraint types, troubleshooting, and the full threat model.
• Tenuo Core Concepts
• Security Model
• Example Code