An agent that can delete a production table needs a human to say yes first. The yes is the easy part. The hard part is the wait before it: the approver closes the tab, the worker restarts during a deploy, and the click lands on a process that no longer remembers what it was approving.
The first implementation is usually a Python dict of pending requests and an asyncio.sleep loop. It works in a demo. Then the worker restarts and the request disappears, the approval arrives twice and the action runs twice, or nobody answers and the coroutine holds a connection until the pod gets evicted.
Temporal handles the waiting part at the infrastructure level. A Workflow is a state machine rebuilt from an event history, so a worker can die mid-wait and another one continues from the same point. workflow.wait_condition waits for days without holding a thread. Signals carry the decision. Timers survive restarts. This tutorial builds that approval gate in front of a destructive action, then breaks it on purpose to check the failure cases.
I ran every command below against Temporal CLI 1.10.0 (server 1.32.2) and temporalio 1.35.0. The output blocks are copied from those runs.
What you need
- Python 3.10 or newer
- Temporal CLI
temporalioandopenaifrom PyPI- Optional:
OPENAI_API_KEY. Without it, the demo classifies risk with a keyword heuristic, so the whole thing runs offline. - 20 minutes
Step 1: install the CLI and start the dev server
Grab the CLI archive for your platform. This is the Linux amd64 build:
curl -sSfL "https://temporal.download/cli/archive/latest?platform=linux&arch=amd64" -o temporal.tar.gz
tar xzf temporal.tar.gz
sudo mv temporal /usr/local/bin
temporal --version
Set up a project and install the SDK:
mkdir approval-agent && cd approval-agent
python3 -m venv .venv
source .venv/bin/activate
pip install temporalio openai
Start the dev server in its own terminal and leave it running:
temporal server start-dev
The server listens on localhost:7233. The Web UI is at http://localhost:8233, which is where you will watch the workflow history later.
Step 2: define the data that crosses the boundary
Three dataclasses. Keep them boring and serializable, because every field ends up in the event history.
# models.py
from dataclasses import dataclass
@dataclass
class AgentRequest:
prompt: str
timeout_seconds: int = 300
@dataclass
class ActionProposal:
action: str
risk: str
rationale: str
@dataclass
class ApprovalDecision:
request_id: str
approved: bool
reviewer: str
reason: str = ""
decided_at: str = ""
Step 3: put the model call in an activity
Activities are where side effects live: HTTP calls, database writes, shell commands. Temporal retries them on failure, so they should be idempotent.
The proposal activity calls the Responses API with a JSON schema when a key exists, and falls back to a keyword check when it does not. That fallback is not a toy. It lets you test the approval flow without burning tokens, and it keeps the tutorial runnable on a plane.
# activities.py
import json
import os
from temporalio import activity
from models import ActionProposal, AgentRequest
SYSTEM_PROMPT = """You classify an infrastructure request before an agent acts on it.
Return a short action, a risk level (low or high), and one sentence explaining why.
Mark anything that deletes, drops, restarts, or touches production as high risk."""
SCHEMA = {
"type": "object",
"properties": {
"action": {"type": "string"},
"risk": {"type": "string", "enum": ["low", "high"]},
"rationale": {"type": "string"},
},
"required": ["action", "risk", "rationale"],
"additionalProperties": False,
}
RISKY_WORDS = ("delete", "drop", "truncate", "restart", "prod", "rm -rf", "revoke")
def _offline_proposal(prompt: str) -> ActionProposal:
lowered = prompt.lower()
risk = "high" if any(word in lowered for word in RISKY_WORDS) else "low"
return ActionProposal(
action=prompt.strip(),
risk=risk,
rationale="Offline heuristic: matched a destructive keyword."
if risk == "high"
else "Offline heuristic: no destructive keyword found.",
)
@activity.defn
async def propose_action(req: AgentRequest) -> ActionProposal:
if not os.getenv("OPENAI_API_KEY"):
activity.logger.info("no OPENAI_API_KEY, using offline heuristic")
return _offline_proposal(req.prompt)
from openai import AsyncOpenAI
client = AsyncOpenAI()
resp = await client.responses.create(
model=os.getenv("AGENT_MODEL", "gpt-6-luna"),
input=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": req.prompt},
],
text={
"format": {
"type": "json_schema",
"name": "action_proposal",
"strict": True,
"schema": SCHEMA,
}
},
)
return ActionProposal(**json.loads(resp.output_text))
@activity.defn
async def notify_approval_needed(request_id: str, proposal: ActionProposal) -> None:
activity.logger.info(
"APPROVAL NEEDED request_id=%s action=%s why=%s",
request_id,
proposal.action,
proposal.rationale,
)
@activity.defn
async def execute_action(request_id: str, proposal: ActionProposal) -> str:
# Real tools go here. Use request_id as the idempotency key.
activity.logger.info("EXECUTING request_id=%s action=%s", request_id, proposal.action)
return f"executed: {proposal.action}"
notify_approval_needed is where the Slack message or email goes. In this sample it writes to the worker log, which is enough to see the ordering.
Step 4: write the workflow
This is the part that survives crashes. The workflow owns the request ID, the wait, and the decision.
# workflows.py
import asyncio
from datetime import timedelta
from temporalio import workflow
from temporalio.common import RetryPolicy
with workflow.unsafe.imports_passed_through():
from activities import execute_action, notify_approval_needed, propose_action
from models import ActionProposal, AgentRequest, ApprovalDecision
@workflow.defn
class ApprovalAgentWorkflow:
def __init__(self) -> None:
self._decision: ApprovalDecision | None = None
self._request_id: str | None = None
self._state = "new"
@workflow.signal
def approval_decision(self, decision: ApprovalDecision) -> None:
if decision.request_id != self._request_id:
workflow.logger.warning(
"ignoring decision for stale request %s (current %s)",
decision.request_id,
self._request_id,
)
return
if self._decision is None:
self._decision = decision
@workflow.query
def status(self) -> dict:
return {"state": self._state, "request_id": self._request_id}
@workflow.run
async def run(self, req: AgentRequest) -> dict:
self._request_id = str(workflow.uuid4())
retry = RetryPolicy(maximum_attempts=3, initial_interval=timedelta(seconds=2))
self._state = "analyzing"
proposal: ActionProposal = await workflow.execute_activity(
propose_action,
req,
start_to_close_timeout=timedelta(seconds=60),
retry_policy=retry,
)
if proposal.risk == "low":
self._state = "executing"
output = await workflow.execute_activity(
execute_action,
args=[self._request_id, proposal],
start_to_close_timeout=timedelta(seconds=60),
retry_policy=retry,
)
self._state = "done"
return {
"outcome": "auto_approved",
"request_id": self._request_id,
"action": proposal.action,
"result": output,
}
self._state = "awaiting_approval"
await workflow.execute_activity(
notify_approval_needed,
args=[self._request_id, proposal],
start_to_close_timeout=timedelta(seconds=30),
)
try:
await workflow.wait_condition(
lambda: self._decision is not None,
timeout=timedelta(seconds=req.timeout_seconds),
)
except asyncio.TimeoutError:
self._state = "timed_out"
return {
"outcome": "timed_out",
"request_id": self._request_id,
"action": proposal.action,
"reviewer": None,
"reason": f"no decision within {req.timeout_seconds}s",
}
decision = self._decision
if not decision.approved:
self._state = "rejected"
return {
"outcome": "rejected",
"request_id": self._request_id,
"action": proposal.action,
"reviewer": decision.reviewer,
"reason": decision.reason,
}
self._state = "executing"
output = await workflow.execute_activity(
execute_action,
args=[self._request_id, proposal],
start_to_close_timeout=timedelta(seconds=60),
retry_policy=retry,
)
self._state = "done"
return {
"outcome": "approved_and_executed",
"request_id": self._request_id,
"action": proposal.action,
"reviewer": decision.reviewer,
"reason": decision.reason,
"decided_at": decision.decided_at,
"result": output,
}
Four details matter here.
str(workflow.uuid4()) keeps the ID deterministic while making it a string. I left the cast out the first time and spent a while debugging an approval that was silently ignored. More on that below.
The signal handler checks request_id before storing anything. A late click from an old approval page cannot approve a new action.
if self._decision is None makes the first decision final. Duplicate clicks do not run the action twice.
workflow.wait_condition with a timeout turns silence into a defined outcome. No approval means no execution, and the result says why.
Step 5: worker and starter
The worker registers the workflow and the three activities on one task queue.
# worker.py
import asyncio
import logging
from temporalio import workflow
from temporalio.client import Client
from temporalio.worker import Worker
with workflow.unsafe.imports_passed_through():
from activities import execute_action, notify_approval_needed, propose_action
from workflows import ApprovalAgentWorkflow
TASK_QUEUE = "approval-agents"
async def main() -> None:
logging.basicConfig(level=logging.INFO)
client = await Client.connect("localhost:7233")
worker = Worker(
client,
task_queue=TASK_QUEUE,
workflows=[ApprovalAgentWorkflow],
activities=[propose_action, notify_approval_needed, execute_action],
)
print(f"worker polling {TASK_QUEUE}")
await worker.run()
if __name__ == "__main__":
asyncio.run(main())
# starter.py
import argparse
import asyncio
import uuid
from temporalio import workflow
from temporalio.client import Client
with workflow.unsafe.imports_passed_through():
from models import AgentRequest
from workflows import ApprovalAgentWorkflow
TASK_QUEUE = "approval-agents"
async def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("prompt")
parser.add_argument("--timeout", type=int, default=300)
parser.add_argument("--wait", action="store_true")
args = parser.parse_args()
client = await Client.connect("localhost:7233")
handle = await client.start_workflow(
ApprovalAgentWorkflow.run,
AgentRequest(prompt=args.prompt, timeout_seconds=args.timeout),
id=f"approval-{uuid.uuid4()}",
task_queue=TASK_QUEUE,
)
print(f"workflow_id={handle.id}")
if args.wait:
print(f"result={await handle.result()}")
if __name__ == "__main__":
asyncio.run(main())
Start the worker in one terminal:
python worker.py
Step 6: run it
In a second terminal, start a request that should never be auto-approved:
python starter.py "Delete all test data from the production database" --timeout 600
workflow_id=approval-ff8267d6-7a1d-495f-b819-026c0107d234
The worker logs the approval request, then goes quiet. The workflow is not polling. It is parked, waiting for an event, and it costs nothing while parked.
Check the state from any client, which is what a UI would do:
temporal workflow query --workflow-id "$WF" --type status
Query result:
QueryResult {"request_id":"dda95a8b-b256-42b6-814b-6ace48b942ee","state":"awaiting_approval"}
Send the approval with the helper script:
python send_approval.py "$WF" "$RID" approve "checked the table name" --reviewer arya
Or send it from anything that can call the CLI, which is how a Slack bot or CI job would do it:
temporal workflow signal --workflow-id "$WF" --name approval_decision \
--input '{"request_id":"<request-id>","approved":true,"reviewer":"cli","reason":"on-call checked"}'
Fetch the final result:
temporal workflow result --workflow-id "$WF"
Results:
Status COMPLETED
Result {"action":"Delete all test data from the production database","decided_at":"2026-11-04T05:06:44.299207+00:00","outcome":"approved_and_executed","reason":"checked table name","request_id":"dda95a8b-b256-42b6-814b-6ace48b942ee","result":"executed: Delete all test data from the production database","reviewer":"arya"}
Try a low-risk request and the workflow skips the wait:
result={'action': 'Check disk usage on web-01', 'outcome': 'auto_approved', 'request_id': '0ad4281f-3989-4b82-97b3-6151e5f6b567', 'result': 'executed: Check disk usage on web-01'}
Let the timer expire and the action never runs:
result={'action': 'Revoke old deploy keys on staging', 'outcome': 'timed_out', 'reason': 'no decision within 5s', 'request_id': '7b991626-98bd-4365-ba0b-219afb60dc75', 'reviewer': None}
Step 7: kill the worker while it waits
This is the reason to use Temporal instead of a dict. Start a request, confirm it is waiting, then kill the worker process outright:
kill <worker-pid>
The workflow stays open. The state lives in the Temporal service history, not in the dead process. A query still answers:
QueryResult {"request_id":"e938f3ea-95ff-4f80-a1a2-619595dbbdec","state":"awaiting_approval"}
Start a new worker, send the approval, and it completes with the original request ID:
Results:
Status COMPLETED
Result {"action":"Restart the production postgres primary","decided_at":"2026-11-04T05:07:10.866080+00:00","outcome":"approved_and_executed","reason":"approved after worker restart","request_id":"e938f3ea-95ff-4f80-a1a2-619595dbbdec","result":"executed: Restart the production postgres primary","reviewer":"arya"}
There is no recovery code to write, and no table of pending approvals to keep in sync. The history is the state.
Step 8: the bug that made approvals disappear
My first version used this line:
self._request_id = workflow.uuid4()
The signal carried a string from JSON. The workflow held a uuid.UUID object. != returned true, so every decision was dropped, and the warning printed the same value twice:
ignoring decision for stale request 5b1b1df2-b092-4933-ae78-621bce2e1f9f (current 5b1b1df2-b092-4933-ae78-621bce2e1f9f)
The log looks like a broken comparison because it is. workflow.uuid4() returns a uuid.UUID in temporalio. Cast it to str at the boundary, or parse the signal payload into a UUID before comparing. Pick one and be consistent for every ID that crosses into the workflow.
Step 9: what to harden before production
- Authenticate the sender. A signal is just an API call. Put the approval endpoint behind your auth layer and keep the Temporal namespace private.
- Use
request_idas the idempotency key for the real tool. If the activity retries after a timeout, the tool should recognize the same key. - Log the decision fields in the result, not just "approved". Reviewer, timestamp, and reason are what an audit asks for.
- Keep the timeout in the input, not hardcoded. Different actions deserve different patience.
- Add a query for the pending action text so the UI can show what is being approved, not just an ID.
When a database table is enough
If you have one approval step, one worker, and a Postgres instance already, a table with status, request_id, and a cron poller is simpler than running another service. Temporal earns its keep when the workflow has multiple steps, long waits, retries around flaky tools, or an audit requirement that says the exact sequence must be replayable.
LangGraph's interrupt plus a checkpointer does something similar inside a graph. If your agent is already a LangGraph graph, use that. If the approval gate sits between heterogeneous services, shell jobs, and LLM calls, Temporal keeps the orchestration outside the model code where it is easier to test.
Next steps
Wire the signal into the place where your team already lives. A Slack button handler that calls handle.signal("approval_decision", ...) with the request ID is about thirty lines. Add a query for the proposal text and the message can show what will run. When that works, the same workflow runs against Temporal Cloud by changing the client address and credentials, no code changes.
References
- Human-in-the-loop AI agent recipe: https://docs.temporal.io/ai/cookbook/human-in-the-loop-python
- Workflow message passing (signals, queries, updates): https://docs.temporal.io/develop/python/message-passing
- Set up a local Temporal Service with the Python SDK: https://docs.temporal.io/develop/python/set-up-your-local-python
- OpenAI structured outputs: https://developers.openai.com/api/docs/guides/structured-outputs