continuum-temporal

Continuum Temporal Skill

Authoritative sources: docs/temporal/getting-started.md, workflow-patterns.md, custom-agents.md, custom-workflows.md, human-in-loop.md, docker.md.

---

Setup

pip install "shyftlabs-continuum[temporal]"
docker compose --profile temporal up -d

TEMPORAL_ENABLED=true
TEMPORAL_HOST=localhost:7233
TEMPORAL_NAMESPACE=default
TEMPORAL_TASK_QUEUE=orchestrator-agents
TEMPORAL_ENABLE_HUMAN_IN_LOOP=true

---

Three actors

1. Worker — long-running process; polls task queue, executes activities.
2. Workflow — declarative durable plan; lives inside Temporal.
3. Caller — your app; submits workflows, signals/queries them.

---

Imports

from continuum.temporal import (
    get_temporal_client, get_worker_manager, get_agent_registry,
    WorkflowInput, WorkflowResult,
    AgentStep, ApprovalStep, ParallelStep, ConditionalStep, WaitStep,
    AgentWorkflow, SequentialAgentWorkflow,
    ParallelAgentWorkflow, LoopAgentWorkflow,
    ApprovalDecision, ApprovalRequest,
    HumanInLoopManager, ApprovalNotificationConfig, NotificationParams,
)

---

Hello-Temporal

async def main():
    registry = get_agent_registry()
    registry.register(BaseAgent(name="summarizer", instructions="...", model="gpt-4o-mini"))

    client = get_temporal_client()
    await client.connect()
    await get_worker_manager(client, registry).start()         # blocks

    handle = await client.run_agent_workflow(
        WorkflowInput(
            steps=[
                {"type": "agent",    "agent_name": "summarizer"},
                {"type": "approval", "description": "Review", "approvers": ["alice"]},
                {"type": "agent",    "agent_name": "summarizer"},
            ],
            initial_input="...",
        ),
        id="hello-001",
    )
    result = await handle.result()
    print(result.status, result.content)