Temporal SDK Integration

Complete reference for integrating Predicate Authority with your Temporal.io workers.

Installation

pip install predicate-temporal

Prerequisites

The Predicate Authority Sidecar must be running. See Sidecar and Operations for detailed setup.

Quick start:

# Docker (recommended)
docker run -d -p 8787:8787 ghcr.io/predicatesystems/predicate-authorityd:latest

# Or download binary
curl -fsSL https://github.com/PredicateSystems/predicate-authority-sidecar/releases/latest/download/predicate-authorityd-darwin-arm64.tar.gz | tar -xz
./predicate-authorityd --port 8787 --policy-file policy.json

Quick Start

from temporalio.worker import Worker
from predicate_temporal import PredicateInterceptor
from predicate_authority import AuthorityClient

# Initialize the Predicate Authority client
ctx = AuthorityClient.from_env()

# Create the interceptor
interceptor = PredicateInterceptor(
    authority_client=

How It Works

The interceptor sits in the Temporal activity execution pipeline:

1. Temporal dispatches an activity to your worker
2. Before the activity code runs, the interceptor extracts:
   • Activity name (used as action)
   • Activity arguments (available for context)
3. The interceptor calls AuthorityClient.authorize() to request a mandate
4. If denied: raises an error — activity never executes
5. If approved: activity proceeds normally

This ensures that no untrusted code or payload reaches your OS until it has been cryptographically authorized.

Configuration

Environment Variables

# Required for AuthorityClient.from_env()
export PREDICATE_AUTHORITY_POLICY_FILE=/path/to/policy.json
export PREDICATE_AUTHORITY_SIGNING_KEY=your-secret-key
export PREDICATE_AUTHORITY_MANDATE_TTL_SECONDS=300

Interceptor Options

from predicate_temporal import PredicateInterceptor
from predicate_authority import AuthorityClient

interceptor = PredicateInterceptor(
    # Required: The Predicate Authority client instance
    authority_client=AuthorityClient.from_env().client,

    # Optional: Principal ID (default: "temporal-worker")
    principal="my-worker",

API Reference

Interceptor Parameters

Error Handling

When authorization is denied, the interceptor raises an error that propagates to the workflow:

from temporalio.exceptions import ActivityError, ApplicationError

try:
    await workflow.execute_activity(
        dangerous_activity,
        args,
        start_to_close_timeout=timedelta(seconds=30),
    )

Policy Configuration

Create a policy file that defines allowed and denied activities:

{
  "rules": [
    {
      "name": "allow-safe-activities",
      "effect": "allow",
      "principals": ["temporal-worker"],
      "actions": ["process_order", "send_notification", "fetch_data"],
      "resources": ["*"]
    },
    {
      "name": "deny-dangerous-activities",
      "effect": "deny",
      "principals": ["*"],
      "actions": ["delete_*", "drop_*", "admin_*"],
      "resources": ["*"]
    }
  ]
}

Starter policy packs: The sidecar repository includes ready-to-use policy templates for common use cases including Temporal workers, database access, and file system operations.

Policy Rule Fields

Complete Example

Here's a full example with workflow, activities, and interceptor:

# activities.py
from temporalio import activity

@activity.defn
async def process_order(order_id: str) -> dict:
    """Safe activity - allowed by policy."""
    return

Audit and Compliance

All authorization decisions are logged by the sidecar. For production deployments, connect to the Predicate Control Plane for:

• Global kill-switches — Instantly revoke compromised principals
• Immutable audit vault — 7-year WORM-ready retention
• SIEM integrations — Stream to Datadog, Splunk, or Sentinel
• Fleet management — Centralized policy across all workers

See Audit Logging & Provenance for configuration details.

Source Code