OpenBox SDK provides governance and observability for Temporal workflows by capturing workflow/activity lifecycle events, HTTP telemetry, database queries, and file operations, then sending them to OpenBox Core for policy evaluation.

Key Features:

• 6 event types (WorkflowStarted, WorkflowCompleted, WorkflowFailed, SignalReceived, ActivityStarted, ActivityCompleted)
• 5-tier verdict system (ALLOW, CONSTRAIN, REQUIRE_APPROVAL, BLOCK, HALT)
• Hook-level governance — per-operation evaluation (HTTP requests, file I/O, database queries, function tracing) with started/completed stages
• HTTP/Database/File I/O instrumentation via OpenTelemetry
• Guardrails: Input/output validation and redaction
• Human-in-the-loop approval with expiration handling
• Zero-code setup via create_openbox_worker() factory

pip install openbox-temporal-sdk-python

Requirements:

• Python 3.11+
• Temporal SDK 1.23+ (1.8+ for factory-only usage)
• OpenTelemetry API/SDK 1.38.0+

Use OpenBoxPlugin for drop-in integration with Temporal Workers:

import os
from temporalio.worker import Worker
from openbox.plugin import OpenBoxPlugin

worker = Worker(
    client,
    task_queue="my-task-queue",
    workflows=[MyWorkflow],
    activities=[my_activity],
    plugins=[
        OpenBoxPlugin(
            openbox_url=os.getenv("OPENBOX_URL"),
            openbox_api_key=os.getenv("OPENBOX_API_KEY"),
        )
    ],
)

await worker.run()

The plugin automatically configures governance interceptors, OTel instrumentation, sandbox passthrough, W3C trace propagation through Temporal headers (via temporalio.contrib.opentelemetry.TracingInterceptor), and the send_governance_event activity.

Credentials never leave the plugin. openbox_api_key is captured on the governance activity instance itself — it does not flow through activity inputs, so it is never written to workflow history. To opt out of trace propagation (e.g., if you already wire OpenTelemetryPlugin), pass enable_trace_propagation=False.

from temporalio.contrib.opentelemetry import OpenTelemetryPlugin

worker = Worker(
    client,
    task_queue="my-task-queue",
    workflows=[MyWorkflow],
    activities=[my_activity],
    plugins=[
        OpenTelemetryPlugin(),
        OpenBoxPlugin(openbox_url=..., openbox_api_key=...),
    ],
)

Requires temporalio >= 1.23.0. For older versions, use create_openbox_worker() below.

Use the create_openbox_worker() factory for simple integration:

import os
from openbox import create_openbox_worker

worker = create_openbox_worker(
    client=client,
    task_queue="my-task-queue",
    workflows=[MyWorkflow],
    activities=[my_activity],
    # OpenBox config
    openbox_url=os.getenv("OPENBOX_URL"),
    openbox_api_key=os.getenv("OPENBOX_API_KEY"),
)

await worker.run()

The factory automatically:

1. Validates the API key
2. Creates span processor
3. Sets up OpenTelemetry instrumentation
4. Creates governance interceptors (incl. W3C trace propagation)
5. Builds the GovernanceActivities instance with credentials captured on self and registers its send_governance_event method — the API key is never passed through activity inputs / workflow history
6. Returns fully configured Worker

OPENBOX_URL=http://localhost:8086
OPENBOX_API_KEY=obx_test_key_1
OPENBOX_GOVERNANCE_TIMEOUT=30.0
OPENBOX_GOVERNANCE_POLICY=fail_open  # or fail_closed

worker = create_openbox_worker(
    client=client,
    task_queue="my-task-queue",
    workflows=[MyWorkflow],
    activities=[my_activity],

    # OpenBox config
    openbox_url="http://localhost:8086",
    openbox_api_key="obx_test_key_1",
    governance_timeout=30.0,
    governance_policy="fail_open",

    # Event filtering
    send_start_event=True,
    send_activity_start_event=True,
    skip_workflow_types={"InternalWorkflow"},
    skip_activity_types={"send_governance_event"},
    skip_signals={"heartbeat"},

    # Database instrumentation
    instrument_databases=True,
    db_libraries={"psycopg2", "sqlalchemy"},  # None = all available
    sqlalchemy_engine=engine,  # pass pre-existing engine for query capture

    # File I/O instrumentation
    instrument_file_io=False,  # disabled by default

    # Header-based W3C trace propagation (client → workflow → activities).
    # Default True. Set False if you already wire OpenTelemetryPlugin or a
    # custom propagator.
    enable_trace_propagation=True,

    # Standard Worker options (all supported)
    activity_executor=my_executor,
    max_concurrent_activities=10,
)

OpenBox Core returns a verdict indicating what action the SDK should take.

v1.0 Backward Compatibility:

• "continue" → ALLOW
• "stop" → HALT
• "require-approval" → REQUIRE_APPROVAL

OpenBox Core can validate and redact sensitive data before/after activity execution:

# Request
{
  "verdict": "allow",
  "guardrails_result": {
    "input_type": "activity_input",
    "redacted_input": {"prompt": "[REDACTED]", "user_id": "123"},
    "validation_passed": true,
    "reasons": []
  }
}

# If validation fails:
{
  "validation_passed": false,
  "reasons": [
    {"type": "pii", "field": "email", "reason": "Contains PII"}
  ]
}

Configure error policy via on_api_error (constants available as hook_governance.FAIL_OPEN / FAIL_CLOSED):

• httpx (sync + async) - full body capture
• requests - full body capture
• urllib3 - full body capture
• urllib - request body only

Fully supported — any dbapi-compatible library using OTel's CursorTracer.traced_execution():

• psycopg2, pymysql, mysql-connector-python, and other dbapi-compliant drivers

Custom hooks (best-effort) — these libraries use non-dbapi instrumentation paths; governance hooks may not work correctly in all scenarios:

• asyncpg — wrapt wrapper on Connection methods (governance runs outside OTel span context)
• pymongo — CommandListener monitoring + wrapt Collection wrappers (dedup via thread-local flag; some internal commands like endSessions only produce completed stage)
• redis — native OTel request_hook/response_hook
• sqlalchemy — before/after_cursor_execute event listeners

SQLAlchemy Note: If your SQLAlchemy engine is created before create_openbox_worker() runs (e.g., at module import time), you must pass it via the sqlalchemy_engine parameter. Without this, SQLAlchemyInstrumentor only patches future create_engine() calls and won't capture queries on pre-existing engines.

from db.engine import engine

worker = create_openbox_worker(
    ...,
    db_libraries={"psycopg2", "sqlalchemy"},
    sqlalchemy_engine=engine,
)

• open(), read(), write(), readline(), readlines()
• Skips system paths (/dev/, /proc/, /sys/, __pycache__)

Every HTTP request, file operation, and database query made during an activity is evaluated by OpenBox Core in real-time at two stages:

Per-operation governance evaluates every read()/write()/readline()/readlines()/writelines() call, not just open/close:

How it works (HTTP):

1. OTel httpx instrumentation fires a request hook → SDK sends started governance evaluation with request data
2. If verdict is BLOCK/HALT → request is aborted before it leaves the process
3. After response arrives → SDK sends completed governance evaluation with full request+response data
4. If verdict is BLOCK/HALT → GovernanceBlockedError is raised, activity fails with GovernanceStop

How it works (File I/O):

1. Activity calls open() → SDK sends started governance evaluation with file path and mode
2. If verdict is BLOCK/HALT → file is never opened, GovernanceBlockedError is raised
3. Each read()/write()/readline()/readlines()/writelines() call sends started (before) and completed (after) governance evaluations — enabling content-based policy enforcement
4. After file is closed → SDK sends completed governance with lifecycle summary (total bytes, operations list)
5. File governance requires instrument_file_io=True (disabled by default)

A simple open-read-close produces 4 governance evaluations: open(started) → read(started) → read(completed) → close(completed).

Every database operation is evaluated at started (pre-query, can block) and completed (post-query, reports outcome):

How it works:

1. Activity executes a DB query (via any supported library)
2. SDK governance hook intercepts before the query → sends started evaluation
3. If verdict is BLOCK/HALT → query is aborted, GovernanceBlockedError raised
4. Query executes normally → SDK sends completed evaluation with duration and error status
5. DB governance is automatic when instrument_databases=True (default)

Per-library strategy:

Note: Libraries marked "Fully supported" use OTel's CursorTracer, which guarantees governance hooks run inside the OTel span context. Best-effort libraries use custom hooks that may produce inconsistencies (e.g., missing stages for internal commands). Some C extension types (e.g., psycopg2.extensions.cursor) cannot be patched with wrapt — in those cases governance hooks are silently skipped, but OTel span capture still works normally.

Functions decorated with @traced are automatically governed when hook_governance is configured:

How it works (Function Tracing):

1. @traced decorator wrapper starts OTel span
2. → SDK sends started governance evaluation with function name and module
3. If verdict is BLOCK/HALT → GovernanceBlockedError raised, function never executes
4. Function executes normally
5. → SDK sends completed governance evaluation with result or error info
6. If verdict is BLOCK/HALT → GovernanceBlockedError raised after execution
7. When hook_governance is not configured → zero overhead, no governance calls

Governed span tracking: When hook-level governance is active, the SDK marks HTTP spans as "governed" so the OTel on_end processor skips buffering them — preventing duplicate spans.

See System Architecture for detailed component design.

High-Level Flow:

Workflow/Activity → Interceptors → Span Processor → OpenBox Core API
                                                    ↓
                                            Returns Verdict
                                                    ↓
                                    (ALLOW, BLOCK, HALT, etc.)

Hook-Level (per HTTP request):
Activity HTTP Call → OTel Hook → hook_governance → API (started) → Allow/Block
                   → Response  → hook_governance → API (completed) → Allow/Block

Hook-Level (per file operation):
Activity open()       → hook_governance → API (started)   → Allow/Block
Activity read/write() → hook_governance → API (started)   → Allow/Block
                      → hook_governance → API (completed) → Allow/Block
Activity close()      → hook_governance → API (completed) → Allow/Block

Hook-Level (per DB query):
Activity DB Call → db_governance_hooks → hook_governance → API (started) → Allow/Block
                → Query executes     → hook_governance → API (completed)

Hook-Level (per @traced function):
@traced call    → hook_governance → API (started) → Allow/Block
                → Function runs  → hook_governance → API (completed)

Module responsibilities:

• otel_setup.py — OTel instrumentation, hooks, TracedFile wrapper
• hook_governance.py — Shared governance evaluator (payload building, API calls, verdict handling)
• db_governance_hooks.py — Per-library DB governance wrappers (wrapt, OTel hooks, SQLAlchemy events)
• span_processor.py — Span buffering, activity context tracking
• activity_interceptor.py — Activity lifecycle governance events

For manual control, import individual components:

from openbox import (
    initialize,
    WorkflowSpanProcessor,
    GovernanceInterceptor,
    GovernanceConfig,
)
from openbox.otel_setup import setup_opentelemetry_for_governance
from openbox.activity_interceptor import ActivityGovernanceInterceptor
from openbox.activities import build_governance_activities
from openbox.hook_governance import FAIL_OPEN, FAIL_CLOSED  # error policy constants

# 1. Initialize SDK
initialize(api_url="http://localhost:8086", api_key="obx_test_key_1")

# 2. Create span processor
span_processor = WorkflowSpanProcessor(
    ignored_url_prefixes=["http://localhost:8086"]
)

# 3. Setup OTel instrumentation (governance always enabled)
setup_opentelemetry_for_governance(
    span_processor,
    api_url="http://localhost:8086",
    api_key="obx_test_key_1",
    sqlalchemy_engine=engine,  # optional: instrument pre-existing engine
)

# 4. Create governance config
config = GovernanceConfig(
    on_api_error="fail_closed",
    api_timeout=30.0,
)

# 5. Create interceptors
workflow_interceptor = GovernanceInterceptor(
    api_url="http://localhost:8086",
    api_key="obx_test_key_1",
    span_processor=span_processor,
    config=config,
)

activity_interceptor = ActivityGovernanceInterceptor(
    api_url="http://localhost:8086",
    api_key="obx_test_key_1",
    span_processor=span_processor,
    config=config,
)

# 6. Build the governance activity instance (credentials captured on `self`
# so they never flow through workflow history).
governance_activities = build_governance_activities(
    api_url="http://localhost:8086",
    api_key="obx_test_key_1",
)

# 7. Create worker
from temporalio.worker import Worker
from temporalio.contrib.opentelemetry import TracingInterceptor
worker = Worker(
    client=client,
    task_queue="my-task-queue",
    workflows=[MyWorkflow],
    activities=[my_activity, governance_activities.send_governance_event],
    interceptors=[
        workflow_interceptor,
        activity_interceptor,
        TracingInterceptor(),  # W3C header propagation for OTel spans
    ],
)

• Project Overview & PDR - Requirements, features, constraints
• System Architecture - Component design, data flows, security
• Codebase Summary - Code structure and component details
• Code Standards - Coding conventions and best practices
• Project Roadmap - Future enhancements and timeline

The SDK includes comprehensive test coverage with 15 test files:

pytest tests/

Test files: test_activities.py, test_activity_interceptor.py, test_config.py, test_db_governance_hooks.py, test_file_governance_hooks.py, test_otel_hook_pause.py, test_otel_hook_pause_db.py, test_otel_setup.py, test_plugin.py, test_plugin_integration.py, test_span_processor.py, test_tracing.py, test_types.py, test_worker.py, test_workflow_interceptor.py

MIT License - See LICENSE file for details

• Issues: GitHub Issues
• Documentation: See ./docs/

Version: 1.2.0 | Last Updated: 2026-04-05