Agent Orchestration with Temporal: Durable Execution for Healthcare AI Workflows

Your healthcare AI agent works perfectly in a Jupyter notebook. It calls an LLM, reads FHIR data, generates a prior authorization request, and submits it to the payer. Then the server crashes mid-submission. The LLM call already cost $0.12. The eligibility check already confirmed coverage. The payer received half a request. And nobody knows what state anything is in.

This is the gap between demo-grade and production-grade AI orchestration. Frameworks like LangChain, CrewAI, and AutoGen solve the intelligence problem—how to prompt, chain, and coordinate LLM calls. But they don't solve the infrastructure problem: crash recovery, durable state, long-running human approvals, and compensating transactions when things go wrong halfway through a clinical workflow.

Temporal does. It's a workflow orchestration engine built by the team behind Uber's Cadence, battle-tested at companies like Netflix, Snap, and Stripe. It brings durable execution—the guarantee that your workflow will complete exactly once, even through server crashes, network failures, and deployments—to healthcare AI agent orchestration.

This post shows you how to combine Temporal's infrastructure guarantees with LLM-powered agents to build healthcare workflows that are production-ready from day one. We'll build a complete prior authorization agent, implement saga compensation, add human-in-the-loop approval, and compare Temporal against LangGraph and CrewAI for healthcare use cases.

Why LangChain and CrewAI Orchestration Isn't Enough for Healthcare

If you've built multi-agent systems with LangChain or CrewAI, you know they handle agent coordination well: tool calling, chain-of-thought prompting, multi-agent collaboration. But production healthcare workflows expose four critical gaps:

No Crash Recovery

When a LangChain chain fails mid-execution, the entire state is lost. There's no replay mechanism. If your agent was on step 5 of 8 when the process crashed, you restart from step 1. In healthcare, this means re-running expensive LLM calls, re-querying FHIR servers, and potentially submitting duplicate transactions to payers—a compliance violation under HIPAA and CMS rules.

No Durable State for Long-Running Processes

A prior authorization can take minutes, hours, or days to resolve. A care coordination workflow might span weeks. LangChain and CrewAI maintain state in memory—they're designed for request-response interactions, not workflows that outlive the process that started them. You end up bolting on Redis, Postgres, or a message queue to manage state, building a fragile orchestration layer that Temporal provides out of the box.

No Human Task Management

Healthcare AI systems require human-in-the-loop approval at multiple stages. A clinician must review an AI-generated diagnosis. A pharmacist must approve a medication recommendation. An administrator must authorize a high-cost procedure. LangChain has no native concept of "pause the workflow, wait for a human, resume when approved." You build it yourself with webhooks, polling, and prayer.

No Saga Compensation

When step 6 of an 8-step workflow fails, what happens to the side effects of steps 1 through 5? If you already confirmed eligibility with a payer, submitted a FHIR resource, and notified a care team—and then the final step fails—you need compensating transactions to undo or mitigate the partial work. LangChain and CrewAI have no built-in saga pattern.

Temporal Basics: A Primer for LangChain Developers

If you know LangChain but not Temporal, here's the mental model. Temporal has five core concepts that map loosely to things you already understand:

Workflows are the equivalent of a LangChain chain or agent—they define the sequence of steps. But unlike a chain, a Temporal workflow is durable. Its state is persisted to a database. If the process running it crashes, another process picks it up and continues from exactly where it left off. Workflows are written as regular Python functions (or classes), decorated with @workflow.defn.

Activities are the individual steps—the equivalent of LangChain tools. Each activity is a function that does one thing: call an LLM, query a FHIR server, submit an EDI transaction, or send a notification. Activities can fail, be retried, and time out independently. They're decorated with @activity.defn.

Workers are processes that execute workflows and activities. Think of them as the runtime—similar to how you run agent.invoke(), but workers are long-running processes that pull tasks from a queue. You can scale workers independently: 10 workers for LLM calls, 2 for FHIR operations, 1 for EDI submissions.

Task Queues are named channels that connect workflow requests to workers. When you start a workflow, it goes into a task queue. Workers listening on that queue pick it up. This decouples workflow definition from execution—you can deploy new workflow versions without downtime.

Signals are external messages you can send to a running workflow. This is Temporal's killer feature for healthcare: a workflow can pause and wait for a signal ("clinician approved") indefinitely. The workflow consumes zero resources while waiting. When the signal arrives—hours or days later—the workflow resumes exactly where it paused.

Queries let you inspect a running workflow's state without affecting its execution. "What step is this prior auth on?" "Has the clinician reviewed it yet?" Queries answer these questions in real time.

Healthcare Workflow Example: Prior Authorization Agent

Let's build a prior authorization agent orchestrated by Temporal. The workflow receives a prior auth request, checks eligibility, uses an LLM to extract medical necessity documentation, validates the output against payer rules, submits to the payer via X12 278, waits for a response, and notifies the clinician.

Here's the complete Temporal workflow definition in Python:

import asyncio
from datetime import timedelta
from dataclasses import dataclass
from temporalio import workflow, activity
from temporalio.common import RetryPolicy


@dataclass
class PriorAuthRequest:
    patient_id: str
    procedure_code: str
    provider_npi: str
    payer_id: str
    clinical_notes: str


@dataclass
class PriorAuthResult:
    status: str  # approved, denied, pending_review
    auth_number: str | None
    denial_reason: str | None


@workflow.defn
class PriorAuthWorkflow:
    """Temporal workflow orchestrating a prior auth agent.
    
    Each step is an activity — independently retryable,
    durable across crashes, with full audit history.
    """

    def __init__(self):
        self.payer_response: PriorAuthResult | None = None
        self.current_step = "initialized"

    @workflow.signal
    async def receive_payer_response(self, result: PriorAuthResult):
        """Signal handler — payer sends response asynchronously."""
        self.payer_response = result

    @workflow.query
    def get_status(self) -> str:
        """Query handler — check current workflow step."""
        return self.current_step

    @workflow.run
    async def run(self, request: PriorAuthRequest) -> PriorAuthResult:
        retry = RetryPolicy(
            maximum_attempts=3,
            initial_interval=timedelta(seconds=1),
            backoff_coefficient=2.0,
        )

        # Step 1: Check eligibility with payer
        self.current_step = "checking_eligibility"
        eligibility = await workflow.execute_activity(
            check_eligibility,
            args=[request.patient_id, request.payer_id],
            start_to_close_timeout=timedelta(seconds=10),
            retry_policy=retry,
        )
        if not eligibility["is_eligible"]:
            return PriorAuthResult(
                status="denied",
                auth_number=None,
                denial_reason="Patient not eligible for coverage",
            )

        # Step 2: Extract medical necessity with LLM
        self.current_step = "extracting_medical_necessity"
        medical_necessity = await workflow.execute_activity(
            extract_medical_necessity,
            args=[request.clinical_notes, request.procedure_code],
            start_to_close_timeout=timedelta(seconds=30),
            retry_policy=retry,
        )

        # Step 3: Validate output against payer rules
        self.current_step = "validating_output"
        validation = await workflow.execute_activity(
            validate_prior_auth,
            args=[medical_necessity, request.payer_id],
            start_to_close_timeout=timedelta(seconds=5),
            retry_policy=retry,
        )
        if not validation["is_valid"]:
            return PriorAuthResult(
                status="denied",
                auth_number=None,
                denial_reason=validation["reason"],
            )

        # Step 4: Submit to payer via X12 278
        self.current_step = "submitting_to_payer"
        submission = await workflow.execute_activity(
            submit_to_payer,
            args=[request, medical_necessity],
            start_to_close_timeout=timedelta(seconds=30),
            retry_policy=retry,
        )

        # Step 5: Wait for payer response (signal)
        self.current_step = "waiting_for_payer_response"
        try:
            await workflow.wait_condition(
                lambda: self.payer_response is not None,
                timeout=timedelta(days=14),
            )
        except asyncio.TimeoutError:
            self.payer_response = PriorAuthResult(
                status="pending_review",
                auth_number=None,
                denial_reason="Payer response timeout — escalated",
            )

        # Step 6: Notify clinician
        self.current_step = "notifying_clinician"
        await workflow.execute_activity(
            notify_clinician,
            args=[request.provider_npi, self.payer_response],
            start_to_close_timeout=timedelta(seconds=10),
            retry_policy=retry,
        )

        self.current_step = "completed"
        return self.payer_response

And the activity implementations—the workers that call LLM and FHIR endpoints:

import httpx
import json
from temporalio import activity

FHIR_BASE = "https://ehr.example.com/fhir"
LLM_ENDPOINT = "https://api.openai.com/v1/chat/completions"


@activity.defn
async def check_eligibility(patient_id: str, payer_id: str) -> dict:
    """FHIR Worker: Query payer eligibility via Coverage resource."""
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            f"{FHIR_BASE}/Coverage",
            params={"beneficiary": f"Patient/{patient_id}", "status": "active"},
            headers={"Authorization": f"Bearer {get_fhir_token()}"},
        )
        resp.raise_for_status()
        bundle = resp.json()
        entries = bundle.get("entry", [])
        return {
            "is_eligible": len(entries) > 0,
            "coverage_id": entries[0]["resource"]["id"] if entries else None,
            "payer_id": payer_id,
        }


@activity.defn
async def extract_medical_necessity(
    clinical_notes: str, procedure_code: str
) -> dict:
    """LLM Worker: Use GPT-4o to extract structured medical necessity."""
    prompt = f"""Extract medical necessity documentation for prior authorization.

Procedure Code: {procedure_code}
Clinical Notes:
{clinical_notes}

Return a JSON object with:
- diagnosis_codes: list of relevant ICD-10 codes
- medical_justification: 2-3 sentence clinical justification
- supporting_evidence: list of clinical findings from notes
- urgency: routine | urgent | emergent
"""
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            LLM_ENDPOINT,
            headers={"Authorization": f"Bearer {get_llm_api_key()}"},
            json={
                "model": "gpt-4o",
                "messages": [{"role": "user", "content": prompt}],
                "response_format": {"type": "json_object"},
                "temperature": 0.1,
            },
            timeout=25.0,
        )
        resp.raise_for_status()
        return json.loads(resp.json()["choices"][0]["message"]["content"])


@activity.defn
async def validate_prior_auth(necessity: dict, payer_id: str) -> dict:
    """Rules Worker: Validate LLM output against payer-specific rules."""
    errors = []
    if not necessity.get("diagnosis_codes"):
        errors.append("Missing diagnosis codes")
    if not necessity.get("medical_justification"):
        errors.append("Missing medical justification")
    if len(necessity.get("medical_justification", "")) < 50:
        errors.append("Justification too brief for payer requirements")
    # Payer-specific rule: some payers require >= 2 ICD-10 codes
    if payer_id in STRICT_PAYERS and len(necessity.get("diagnosis_codes", [])) < 2:
        errors.append(f"Payer {payer_id} requires >= 2 diagnosis codes")
    return {"is_valid": len(errors) == 0, "reason": "; ".join(errors)}


@activity.defn
async def submit_to_payer(
    request: "PriorAuthRequest", necessity: dict
) -> dict:
    """EDI Worker: Submit X12 278 prior auth request to clearinghouse."""
    x12_payload = build_x12_278(
        patient_id=request.patient_id,
        procedure_code=request.procedure_code,
        provider_npi=request.provider_npi,
        payer_id=request.payer_id,
        diagnosis_codes=necessity["diagnosis_codes"],
        justification=necessity["medical_justification"],
    )
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            CLEARINGHOUSE_URL,
            content=x12_payload,
            headers={"Content-Type": "application/edi-x12"},
            timeout=20.0,
        )
        resp.raise_for_status()
        return {"tracking_id": resp.headers.get("X-Tracking-ID"), "status": "submitted"}


@activity.defn
async def notify_clinician(provider_npi: str, result: "PriorAuthResult") -> None:
    """Human Worker: Send notification to clinician's review queue."""
    await send_notification(
        recipient_npi=provider_npi,
        subject=f"Prior Auth {result.status.upper()}: {result.auth_number or 'N/A'}",
        body=f"Status: {result.status}\nAuth#: {result.auth_number}\nDetails: {result.denial_reason or 'Approved'}",
        channel="ehr_inbox",  # In-basket message
    )

Finally, the worker process that ties it all together:

import asyncio
from temporalio.client import Client
from temporalio.worker import Worker


async def main():
    client = await Client.connect("localhost:7233")

    worker = Worker(
        client,
        task_queue="healthcare-prior-auth",
        workflows=[PriorAuthWorkflow],
        activities=[
            check_eligibility,
            extract_medical_necessity,
            validate_prior_auth,
            submit_to_payer,
            notify_clinician,
        ],
    )
    await worker.run()


if __name__ == "__main__":
    asyncio.run(main())

Durable Execution: Why Crashes Don't Matter

Here's where Temporal fundamentally differs from every AI agent framework. When you run the prior auth workflow above and the server crashes after step 3 (validate output) but before step 4 (submit to payer), Temporal does something remarkable: it replays the workflow from its event history.

Temporal's server durably stores every event—every activity start, completion, failure, signal, and timer—in an append-only event history. When a worker recovers after a crash, Temporal replays the workflow code against this history. Steps 1, 2, and 3 are not re-executed. Their results are pulled from history. Execution resumes at step 4 as if the crash never happened.

This matters enormously in healthcare for three reasons:

No duplicate submissions. Without Temporal, a crash after submitting to a payer might cause a retry that submits a second request. Duplicate prior auth submissions create confusion, delays, and potential compliance issues. With Temporal, the submission activity is either completed (and its result is in history) or it didn't (and it retries exactly once).

No lost LLM spend. An LLM call that costs $0.10–$2.00 per invocation doesn't need to be re-run if the crash happens after the LLM activity is completed. The result is persisted. Across thousands of workflows, this prevents significant waste—a real concern given the economics of healthcare AI agents.

Complete audit trail. Every event in the workflow history is queryable. Regulators, compliance officers, and clinical auditors can replay exactly what happened, in what order, with what inputs and outputs. This isn't custom logging you build yourself—it's a structural property of Temporal.

Saga Pattern for Compensation: Undoing Partial Work

What happens when step 4 (submit to payer) fails after steps 1–3 have already succeeded? The eligibility check confirmed coverage. The LLM generated medical necessity documentation. The validation passed. But the payer's system is down.

In a naive implementation, you catch the exception and log it. The workflow is stuck in a partially completed state. Someone eventually notices. In a production system, you need the saga pattern—a sequence of compensating transactions that undo or mitigate the effects of previously completed steps.

Here's how to implement it with Temporal:

from dataclasses import dataclass, field
from typing import Callable, Any


@dataclass
class Compensation:
    """Tracks compensating actions for the saga pattern."""
    actions: list = field(default_factory=list)

    def add(self, activity_fn, args: list):
        self.actions.append((activity_fn, args))

    async def compensate(self):
        """Execute compensations in reverse order."""
        for activity_fn, args in reversed(self.actions):
            try:
                await workflow.execute_activity(
                    activity_fn,
                    args=args,
                    start_to_close_timeout=timedelta(seconds=30),
                    retry_policy=RetryPolicy(maximum_attempts=5),
                )
            except Exception as e:
                workflow.logger.error(
                    f"Compensation failed: {activity_fn.__name__}: {e}"
                )


@workflow.defn
class PriorAuthWorkflowWithSaga:
    @workflow.run
    async def run(self, request: PriorAuthRequest) -> PriorAuthResult:
        compensations = Compensation()
        retry = RetryPolicy(maximum_attempts=3)

        try:
            # Step 1: Check eligibility
            eligibility = await workflow.execute_activity(
                check_eligibility,
                args=[request.patient_id, request.payer_id],
                start_to_close_timeout=timedelta(seconds=10),
                retry_policy=retry,
            )
            # Register compensation: release eligibility hold
            compensations.add(release_eligibility_hold, [eligibility["coverage_id"]])

            # Step 2: Extract with LLM
            necessity = await workflow.execute_activity(
                extract_medical_necessity,
                args=[request.clinical_notes, request.procedure_code],
                start_to_close_timeout=timedelta(seconds=30),
                retry_policy=retry,
            )
            # Register compensation: log that extraction was discarded
            compensations.add(log_discarded_extraction, [request.patient_id])

            # Step 3: Validate
            validation = await workflow.execute_activity(
                validate_prior_auth,
                args=[necessity, request.payer_id],
                start_to_close_timeout=timedelta(seconds=5),
                retry_policy=retry,
            )

            # Step 4: Submit to payer
            submission = await workflow.execute_activity(
                submit_to_payer,
                args=[request, necessity],
                start_to_close_timeout=timedelta(seconds=30),
                retry_policy=retry,
            )
            # Register compensation: void the submission
            compensations.add(void_payer_submission, [submission["tracking_id"]])

            return PriorAuthResult(status="submitted", auth_number=None, denial_reason=None)

        except Exception as e:
            workflow.logger.error(f"Workflow failed at step, running compensations: {e}")
            await compensations.compensate()

            # Notify staff about the failure
            await workflow.execute_activity(
                notify_staff_of_failure,
                args=[request.patient_id, str(e)],
                start_to_close_timeout=timedelta(seconds=10),
                retry_policy=retry,
            )

            return PriorAuthResult(
                status="failed",
                auth_number=None,
                denial_reason=f"Workflow failed and compensated: {e}",
            )

The key insight: compensations are registered as each forward step succeeds, then executed in reverse order if any subsequent step fails. Because Temporal durably persists the workflow state, even the compensation logic survives crashes. If the process dies mid-compensation, Temporal replays and continues compensating.

Human-in-the-Loop with Signals

Healthcare AI is never fully autonomous. An LLM might generate a prior auth justification, but a clinician needs to review it before submission. A clinical safety guardrail might flag a drug interaction that requires pharmacist approval. A high-cost procedure might need administrative sign-off.

Temporal signals make human-in-the-loop patterns elegant. The workflow pauses, the human reviews in their own time, and the workflow resumes when the signal arrives:

@dataclass
class HumanReviewDecision:
    approved: bool
    reviewer_id: str
    comments: str
    reviewed_at: str  # ISO 8601 timestamp


@workflow.defn
class PriorAuthWithHumanReview:
    def __init__(self):
        self.review_decision: HumanReviewDecision | None = None

    @workflow.signal
    async def submit_review(self, decision: HumanReviewDecision):
        self.review_decision = decision

    @workflow.query
    def get_review_status(self) -> str:
        if self.review_decision is None:
            return "pending_review"
        return "approved" if self.review_decision.approved else "rejected"

    @workflow.run
    async def run(self, request: PriorAuthRequest) -> PriorAuthResult:
        # ... steps 1-2: eligibility + LLM extraction ...

        # Step 3: Send to human review queue
        await workflow.execute_activity(
            send_to_review_queue,
            args=[request.patient_id, medical_necessity, workflow.info().workflow_id],
            start_to_close_timeout=timedelta(seconds=10),
        )

        # Step 4: WAIT for human review (consumes zero resources)
        try:
            await workflow.wait_condition(
                lambda: self.review_decision is not None,
                timeout=timedelta(hours=4),
            )
        except asyncio.TimeoutError:
            # Escalate if no review in 4 hours
            await workflow.execute_activity(
                escalate_review,
                args=[request.patient_id, workflow.info().workflow_id],
                start_to_close_timeout=timedelta(seconds=10),
            )
            # Wait another 24 hours after escalation
            await workflow.wait_condition(
                lambda: self.review_decision is not None,
                timeout=timedelta(hours=24),
            )

        if self.review_decision and not self.review_decision.approved:
            return PriorAuthResult(
                status="rejected_by_reviewer",
                auth_number=None,
                denial_reason=self.review_decision.comments,
            )

        # Step 5: Submit to payer (only after human approval)
        # ... continue workflow ...

The workflow sits in a waiting_for_signal state consuming zero compute resources. No polling. No cron jobs. No WebSocket connections to maintain. When the clinician clicks "Approve" in the EHR, the review application sends a signal via the Temporal client:

# Called from your review UI's backend
async def approve_prior_auth(workflow_id: str, reviewer: str, comments: str):
    client = await Client.connect("temporal-server:7233")
    handle = client.get_workflow_handle(workflow_id)
    await handle.signal(
        PriorAuthWithHumanReview.submit_review,
        HumanReviewDecision(
            approved=True,
            reviewer_id=reviewer,
            comments=comments,
            reviewed_at=datetime.utcnow().isoformat(),
        ),
    )

This pattern scales to any healthcare approval workflow: medication reviews, discharge planning sign-offs, care plan approvals, billing code reviews. The workflow defines the logic; signals provide the human interaction; Temporal provides the durability.

Temporal vs LangGraph vs CrewAI: Healthcare Comparison

The bottom line: Temporal and LangGraph/CrewAI are not competitors—they operate at different layers. The most robust architecture uses Temporal as the workflow backbone and LangGraph or LangChain within individual activities for complex LLM reasoning. Temporal handles durability, compensation, and human tasks. LangGraph handles multi-step agent reasoning within a single activity.

When to Reach for Temporal

Not every healthcare AI workflow needs Temporal. Use it when your workflow has one or more of these characteristics:

• Multi-step with side effects. If your agent writes to a FHIR server, submits EDI transactions, or sends notifications, you need crash recovery and compensation. A read-only RAG pipeline doesn't.
• Long-running. If the workflow spans more than a single request-response cycle—waiting for payer responses, human approvals, scheduled follow-ups—Temporal's durable timers and signals are essential.
• Compliance-critical. If regulators or auditors need to see exactly what happened, in what order, with what data, Temporal's event history is a structural audit trail, not a logging afterthought.
• High-value. If a failed workflow means a denied claim, a missed diagnosis, or a patient safety issue, the cost of not having durable execution far outweighs the complexity of adopting Temporal.

For simpler use cases—a clinical Q&A bot, a single-shot summarization agent, a CDS Hooks responder—LangChain or a direct API call is perfectly adequate. Don't over-engineer.

Getting Started: From Zero to Temporal

Setting up Temporal locally takes minutes. Here's the quickstart:

# Install Temporal CLI
brew install temporal

# Start Temporal dev server (includes UI at localhost:8233)
temporal server start-dev

# Install Python SDK
pip install temporalio

# Run the worker (from the code above)
python worker.py

# Start a workflow (from a separate script or service)
python -c "
import asyncio
from temporalio.client import Client

async def main():
    client = await Client.connect('localhost:7233')
    result = await client.execute_workflow(
        'PriorAuthWorkflow',
        PriorAuthRequest(
            patient_id='patient-123',
            procedure_code='27447',
            provider_npi='1234567890',
            payer_id='payer-bcbs',
            clinical_notes='Patient presents with severe osteoarthritis...'
        ),
        id='prior-auth-patient-123',
        task_queue='healthcare-prior-auth',
    )
    print(f'Result: {result}')

asyncio.run(main())
"

The Temporal Web UI at localhost:8233 gives you full visibility into running workflows—current step, event history, pending signals, and activity results. For production, Temporal Cloud provides managed hosting, or you can self-host on Kubernetes.

Building Healthcare Agent Infrastructure That Lasts

The healthcare AI landscape is moving fast. Gartner projects that 40% of enterprise applications will embed task-specific AI agents by end of 2026, up from under 5% in 2025. In healthcare, this means AI agents handling prior authorizations, coordinating care transitions, processing claims, managing patient matching, and automating clinical documentation.

The organizations that succeed won't be the ones with the most sophisticated LLM prompts. They'll be the ones whose agent infrastructure can survive a crash at 3 AM without losing a prior auth submission, compensate correctly when a downstream system is unavailable, wait patiently for a clinician to review an AI recommendation, and produce a complete audit trail for every clinical decision.

That's what Temporal gives you. Not smarter agents—more reliable agents.

At Nirmitee, we build healthcare integration and AI infrastructure on foundations designed for production from day one—FHIR-native architectures, durable workflow orchestration, and the full stack from notebook to clinic. If you're building healthcare AI agents that need to work when it matters most, we should talk.

Ready to deploy AI agents in your healthcare workflows? Explore our Agentic AI for Healthcare services to see what autonomous automation can do. We also offer specialized healthcare product engineering services. Talk to our team to get started.

Related reading

For more insights, explore our guides on AI agents in healthcare and FHIR in Modern Healthcare.

You may also find value in EHR Software Development Guide and Building a Healthtech App.