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.

For cross-namespace workflows, Tenuo also supports Temporal Nexus authorization. Nexus lets one team expose a durable service contract through a named Endpoint; Tenuo carries delegated warrants across that Endpoint so the handler can verify the exact operation, input, holder key, expiry, approvals, and revocation state instead of relying only on namespace-level access. For customer-facing scenarios, see Tenuo for Temporal Nexus.

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

If you want the shortest copy-paste path first, start with the Temporal Quickstart. It runs one authorized workflow locally and shows both an allowed Activity and a denied Activity.

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 or signing_key — the holder key the workflow worker uses to sign PoP. The call runs inside the workflow sandbox, so it has to return a key already in memory. signing_key= does that. EnvKeyResolver does too, after preload, and is for development. VaultKeyResolver, AWSSecretsManagerKeyResolver, and GCPSecretManagerKeyResolver start a thread the sandbox blocks, including when their cache is full.

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, pass signing_key= or a resolver whose resolve_sync returns a key already in memory. See the reference. Tenuo Cloud handles key issuance, warrant minting, rotation, and audit for teams that prefer a managed control plane.

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

Use a fresh warrant for one Activity

Long-running workflows can keep a longer-lived issuer warrant as their base authority while issuing a short-lived execution warrant for a consequential dispatch. The override is scoped to this call and is safe when workflow tasks schedule Activities concurrently:

from tenuo.temporal import (
    current_key_id, tenuo_execute_activity, workflow_issue_execution,
)

execution_warrant = await workflow_issue_execution(
    "transfer_funds",
    constraints={"account": account, "amount": amount},
    ttl_seconds=60,
)
await tenuo_execute_activity(
    transfer_funds,
    args=[account, amount],
    warrant=execution_warrant,
    key_id=current_key_id(),
    start_to_close_timeout=timedelta(seconds=30),
)

Scheduled workflows

Use create_scheduled_workflow_with_warrant() so each scheduled workflow start carries Tenuo headers—not warrant material in a memo:

from tenuo.temporal import create_scheduled_workflow_with_warrant

await create_scheduled_workflow_with_warrant(
    client,
    "nightly-report",
    ReportWorkflow.run,
    issuer_warrant,
    "report-agent",
    schedule_spec,
    workflow_args=[tenant_id],
    task_queue="reports",
)

The Schedule action reuses this warrant on every trigger. Its TTL must cover the Schedule lifetime. For unbounded recurring Schedules, use a longer-lived, narrowly scoped issuer warrant and mint short-lived per-Activity execution warrants as shown above.

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.

Cross-namespace calls with Temporal Nexus

When workflows in one Namespace need to invoke capabilities owned by another team or Namespace, use Temporal Nexus authorization. The Nexus helpers mirror the Activity helper shape:

await tenuo_execute_nexus_operation(
    nexus_client,
    PaymentService.refund,
    RefundInput(order_id="ord_123", amount_cents=5000),
    warrant=refund_warrant,
    key_id="agent1",
)

On the handler side, @tenuo_nexus_operation(config, endpoint="payments-prod") or verify_nexus_operation(...) verifies the signed warrant, proof of possession, endpoint/service/operation binding, input constraints, approvals, expiry, and revocation before user code runs.

For workflow-backed Nexus operations, use the explicit envelope/bootstrap helpers documented in the Nexus guide. They carry verified or attenuated authority into the backing workflow without exposing handler Namespace credentials to the caller Namespace.

When Nexus becomes a shared enterprise surface across teams, accounts, or organizations, the remaining challenge is operational control: issuer ownership, trusted-root distribution, revocation rollout, approvals, key rotation, and audit search across the fleet. Tenuo can be self-hosted for those pieces; teams that want that control plane operated centrally can use a managed Tenuo deployment. See Tenuo for Temporal Nexus, or schedule a demo to map this onto your Temporal topology.

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. The workflow worker signs with a holder key it already has in memory (signing_key=, or a KeyResolver whose resolve_sync does not leave the process). EnvKeyResolver is the development form of that, after preload. 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 Quickstart — copy-paste local workflow with one allowed Activity and one denied Activity.
• 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 for Temporal Nexus — practical cross-team, multi-hop, and cross-organization examples.
• Temporal Nexus Authorization — cross-namespace authorization for Nexus operations and workflow-backed handlers.
• Tenuo Core Concepts
• Security Model
• Example Code