Temporal Ai Agent Workflows Durable Execution 2026

Date: 2026-05-10
Track: sandbox-poc
Slug: temporal-ai-agent-workflows-durable-execution-2026
Environment: macOS Darwin 24.6.0, Python 3.12.8

Objective

Validate that the Temporal Python SDK (temporalio) installs cleanly on a stock Python 3.12 environment and that the Workflow + Activity + Signal pattern compiles and imports correctly — establishing a minimal reproducible skeleton for a durable AI agent.

Environment Setup

mkdir -p /tmp/temporal-sandbox
cd /tmp/temporal-sandbox
python3 -m venv venv
source venv/bin/activate
pip install temporalio

Output (truncated):

Successfully installed nexus-rpc-1.4.0 protobuf-6.33.6 temporalio-1.27.0
types-protobuf-6.32.1.20260221 typing-extensions-4.15.0

Installed version: temporalio 1.27.0
Dependencies pulled: nexus-rpc, protobuf, types-protobuf, typing-extensions
Install time: ~8 seconds, no compilation required (pre-built wheels)

Import Validation

from temporalio import workflow, activity
from temporalio.common import RetryPolicy
from temporalio.client import Client
from temporalio.worker import Worker
# Output: All imports OK

Key API surface confirmed available:

• Client.connect() — connects to local dev server or Temporal Cloud
• Client.count_workflows(), Client.create_schedule() — workflow management
• RetryPolicy — configurable exponential backoff with non-retryable error list

Workflow Skeleton — Syntax Validation

File: /tmp/temporal-sandbox/ai_agent_workflow.py

@activity.defn
async def call_llm(prompt: str) -> str:
    return f"[LLM response to: {prompt[:50]}...]"

@activity.defn
async def search_web(query: str) -> str:
    return f"[search results for: {query}]"

@activity.defn
async def persist_checkpoint(step: int, state: dict) -> bool:
    return True

@workflow.defn
class DurableAIAgentWorkflow:
    def __init__(self):
        self._human_approved = False
        self._approval_signal = ""

    @workflow.run
    async def run(self, user_goal: str) -> str:
        retry = RetryPolicy(
            initial_interval=timedelta(seconds=2),
            maximum_attempts=5,
            non_retryable_error_types=["InvalidRequestError"],
        )
        search_result = await workflow.execute_activity(
            search_web, user_goal,
            start_to_close_timeout=timedelta(minutes=1),
            retry_policy=retry,
        )
        plan = await workflow.execute_activity(
            call_llm,
            f"Create an action plan for: {user_goal}\nContext: {search_result}",
            start_to_close_timeout=timedelta(minutes=2),
            retry_policy=retry,
        )
        await workflow.wait_condition(
            lambda: self._human_approved, timeout=timedelta(days=3)
        )
        result = await workflow.execute_activity(
            call_llm,
            f"Execute: {plan}\nApproval: {self._approval_signal}",
            start_to_close_timeout=timedelta(minutes=5),
            retry_policy=retry,
        )
        return result

    @workflow.signal
    async def human_approve(self, notes: str) -> None:
        self._approval_signal = notes
        self._human_approved = True

Run output:

[OK] Workflow class defined: DurableAIAgentWorkflow
[OK] Activities defined: ['call_llm', 'search_web', 'persist_checkpoint']
[OK] Retry policy instantiated — max_attempts: 5
[OK] Signal handler registered: human_approve

Note: Actual execution requires 'temporal server start-dev' (Temporal CLI)
      or Temporal Cloud credentials.

What Worked

• temporalio 1.27.0 installs in one pip command with no compilation needed
• @workflow.defn, @activity.defn, @workflow.signal decorators import and apply correctly
• RetryPolicy instantiates with correct parameters
• workflow.wait_condition() (human-in-the-loop gate that waits up to 3 days) compiles without error
• workflow.execute_activity() pattern with start_to_close_timeout accepted as expected

What Failed / Limitations

• No live execution: Actual workflow runs require either temporal server start-dev (Temporal CLI, installable via brew install temporal) or Temporal Cloud credentials. Neither was available in this sandbox.
• Temporal CLI not pre-installed: Homebrew formula temporal 1.7.0 exists and is installable, but was not installed to avoid persistent changes to the development machine.
• No live event history captured: Because the Temporal server was not started, we cannot demonstrate crash recovery or replay behavior at runtime. The article will clearly state this limitation.

Conclusion

The Python SDK installs cleanly and the core Workflow/Activity/Signal pattern is syntactically valid. The durable AI agent skeleton is a valid starting point for readers. Live execution requires a Temporal server; readers can follow the official temporal server start-dev quickstart to test end-to-end behavior.