“

Series: S12 Developer Tutorials · Article 16 | Level: Intermediate | Reading Time: 15 min

Temporal Workflow Orchestration Guide

#Introduction

In enterprise applications, many business processes involve multiple steps across multiple systems -- approval workflows, data processing pipelines, order fulfillment chains. These long-running processes need persistent state, error retry, timeout handling, and observability. Temporal is a distributed workflow engine designed exactly for this.

coomia-dip uses Temporal as the workflow orchestration infrastructure for Agent Runtime Layer (Agent Runtime). This tutorial guides you through writing your first Temporal workflow with the Python SDK and integrating it into the coomia-dip platform.

#1. Temporal Core Concepts

#1.1 Why a Workflow Engine

Consider implementing a "loan approval" process: receive application, check credit, auto-approve or route to manual review (which may take days), process disbursement, send notifications. Without a workflow engine, you must handle: state recovery after restarts, network timeout retries, long human waits, idempotency for each step. Temporal extracts all this generic complexity from your business code.

#1.2 Core Components

Workflow: Code representation of a business process. Looks like a normal function but with durable execution
Activity: A single step in the workflow. External system interactions belong here
Worker: Process that executes Workflows and Activities by polling Task Queues
Task Queue: Channel for distributing tasks between clients and Workers
Signal: External input to a running Workflow
Query: Read the current state of a running Workflow without affecting it

#1.3 Event Sourcing Under the Hood

Temporal records the complete execution history of every Workflow. When a Worker restarts or a Workflow needs recovery, Temporal replays the event history to reconstruct Workflow state. Your code is re-executed, but Activity results come from history rather than being re-invoked.

This means: code is the state machine, automatic recovery, unlimited waits, and complete audit trails.

#2. Environment Setup

#2.1 Install Temporal Python SDK

Bash

pip install temporalio

#2.2 Verify Connection

Python

from temporalio.client import Client

async def check():
    client = await Client.connect("localhost:7233")
    namespaces = await client.service_client.list_namespaces()
    for ns in namespaces:
        print(f"Namespace: {ns.namespace_info.name}")

#3. Writing Your First Workflow

#3.1 Define Activities

Python

from temporalio import activity
from dataclasses import dataclass

@dataclass
class CreditCheckInput:
    customer_id: str
    application_id: str

@dataclass
class CreditCheckResult:
    score: int
    risk_level: str

@activity.defn
async def check_credit(input: CreditCheckInput) -> CreditCheckResult:
    activity.logger.info(f"Checking credit for {input.customer_id}")
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            "http://credit-service:8080/api/check",
            json={"customer_id": input.customer_id},
        )
        data = resp.json()
    return CreditCheckResult(score=data["score"], risk_level=data["risk_level"])

@activity.defn
async def send_notification(customer_id: str, message: str) -> bool:
    async with httpx.AsyncClient() as client:
        await client.post("http://notif:8080/api/send",
            json={"customer_id": customer_id, "message": message})
    return True

@activity.defn
async def process_disbursement(app_id: str, amount: float) -> str:
    async with httpx.AsyncClient() as client:
        resp = await client.post("http://disburse:8080/api/disburse",
            json={"application_id": app_id, "amount": amount})
    return resp.json()["transaction_id"]

#3.2 Define Workflow

Python

from temporalio import workflow
from datetime import timedelta

@dataclass
class LoanApplication:
    application_id: str
    customer_id: str
    amount: float
    purpose: str

@dataclass
class LoanResult:
    application_id: str
    status: str
    transaction_id: str | None = None
    rejection_reason: str | None = None

@workflow.defn
class LoanApprovalWorkflow:
    def __init__(self):
        self._status = "PENDING"
        self._manual_decision = None

    @workflow.run
    async def run(self, app: LoanApplication) -> LoanResult:
        # Step 1: Credit check
        credit = await workflow.execute_activity(
            check_credit,
            CreditCheckInput(app.customer_id, app.application_id),
            start_to_close_timeout=timedelta(seconds=30),
            retry_policy=RetryPolicy(maximum_attempts=3),
        )

        # Step 2: Decision
        if credit.score >= 700 and app.amount <= 50000:
            self._status = "APPROVED"
        elif credit.score < 500:
            self._status = "REJECTED"
            return LoanResult(app.application_id, "REJECTED",
                rejection_reason=f"Score {credit.score} too low")
        else:
            self._status = "MANUAL_REVIEW"
            await workflow.execute_activity(send_notification,
                args=[app.customer_id, "Under manual review"],
                start_to_close_timeout=timedelta(seconds=10))

            try:
                await workflow.wait_condition(
                    lambda: self._manual_decision is not None,
                    timeout=timedelta(days=7))
            except asyncio.TimeoutError:
                return LoanResult(app.application_id, "REJECTED",
                    rejection_reason="Manual review timeout")

            if self._manual_decision == "REJECTED":
                return LoanResult(app.application_id, "REJECTED",
                    rejection_reason="Rejected by reviewer")
            self._status = "APPROVED"

        # Step 3: Disbursement
        tx_id = await workflow.execute_activity(
            process_disbursement,
            args=[app.application_id, app.amount],
            start_to_close_timeout=timedelta(seconds=60))

        # Step 4: Notify
        await workflow.execute_activity(send_notification,
            args=[app.customer_id, f"Approved! TX: {tx_id}"],
            start_to_close_timeout=timedelta(seconds=10))

        return LoanResult(app.application_id, "APPROVED", tx_id)

    @workflow.signal
    async def manual_decision(self, decision: str, notes: str):
        self._manual_decision = decision

    @workflow.query
    def get_status(self) -> str:
        return self._status

#3.3 Start Worker

Python

async def main():
    client = await Client.connect("localhost:7233")
    worker = Worker(client, task_queue="approval-tasks",
        workflows=[LoanApprovalWorkflow],
        activities=[check_credit, send_notification, process_disbursement])
    await worker.run()

#3.4 Start Workflow

Python

async def start():
    client = await Client.connect("localhost:7233")
    handle = await client.start_workflow(
        LoanApprovalWorkflow.run,
        LoanApplication("LOAN-001", "CUST-123", 30000.0, "expansion"),
        id="loan-LOAN-001", task_queue="approval-tasks")

    status = await handle.query(LoanApprovalWorkflow.get_status)
    print(f"Status: {status}")

    result = await handle.result()
    print(f"Result: {result}")

#4. Platform Integration

#4.1 Trigger Workflows from Actions

Python

@ontology_action(name="submit_loan_application")
async def submit_loan(customer_id: str, amount: float, purpose: str) -> dict:
    client = await Client.connect("localhost:7233")
    app = LoanApplication(f"LOAN-{uuid4().hex[:8]}", customer_id, amount, purpose)
    handle = await client.start_workflow(
        LoanApprovalWorkflow.run, app,
        id=f"loan-{app.application_id}", task_queue="approval-tasks")
    return {"workflow_id": handle.id, "status": "SUBMITTED"}

#4.2 Sync State to Ontology

Python

@activity.defn
async def sync_to_ontology(app_id: str, status: str, details: dict):
    platform = OntoPlatform(base_url="http://localhost:8080", token="svc-token")
    platform.objects.update("LoanApplication", app_id, {"status": status, **details})

#5. Testing

Python

@pytest.mark.asyncio
async def test_auto_approve():
    async with await WorkflowEnvironment.start_time_skipping() as env:
        async with Worker(env.client, task_queue="test",
            workflows=[LoanApprovalWorkflow],
            activities=[check_credit, send_notification, process_disbursement]):
            result = await env.client.execute_workflow(
                LoanApprovalWorkflow.run,
                LoanApplication("T-001", "HIGH-CUST", 20000, "test"),
                id="test-approve", task_queue="test")
            assert result.status == "APPROVED"

#6. Production Best Practices

Workflows must be deterministic: No random(), datetime.now(), or I/O inside Workflows
All side effects in Activities: DB ops, API calls, file I/O must be Activities
Meaningful Workflow IDs: e.g., loan-LOAN-2025-001 for querying and deduplication
Idempotent Activities: Activities may be retried multiple times
Monitor via Temporal Web UI at http://localhost:8233
Alert on: workflow failure rate, task queue backlog, schedule-to-start latency

#Summary

This tutorial covered Temporal workflows in coomia-dip: core concepts, Python SDK programming model, platform integration with Actions and Ontology, testing strategies, and production best practices. Temporal lets you express complex business processes as linear code without manually managing state persistence, error retry, or timeout handling.

Next: [S12-17] OSDK TypeScript Frontend Integration Guide Previous: [S12-15] Flink CDC Real-Time Data Sync Guide