Agent yang boleh menghapus tabel production harus minta izin manusia dulu. Bagian "iya"-nya gampang. Yang susah justru waktu nunggunya: approver nutup tab, worker restart pas deploy, dan kliknya mendarat di proses yang udah nggak inget lagi apa yang mau di-approve.
Implementasi pertama biasanya berupa dict berisi request pending dan loop asyncio.sleep. Jalan di demo. Terus worker restart dan request-nya hilang, approval masuk dua kali dan aksinya jalan dua kali, atau nggak ada yang jawab dan coroutine-nya nahan koneksi sampai pod di-evict.
Temporal ngurus bagian nunggunya di level infrastruktur. Workflow itu state machine yang dibangun ulang dari event history, jadi worker boleh mati di tengah wait dan worker lain lanjut dari titik yang sama. workflow.wait_condition bisa nunggu berhari-hari tanpa nahan thread. Signal yang bawa keputusan. Timer tetap hidup setelah restart. Tutorial ini bikin approval gate di depan aksi destruktif, lalu sengaja dirusak buat ngecek skenario gagalnya.
Semua command di bawah saya jalankan pakai Temporal CLI 1.10.0 (server 1.32.2) dan temporalio 1.35.0. Blok output-nya saya copy langsung dari run tersebut.
Yang kamu butuh
- Python 3.10 atau lebih baru
- Temporal CLI
temporaliodanopenaidari PyPI- Opsional:
OPENAI_API_KEY. Tanpa key, demo pakai heuristic keyword buat nentuin risiko, jadi semuanya bisa jalan offline. - 20 menit
Langkah 1: install CLI dan nyalain dev server
Ambil archive CLI sesuai platform. Ini build Linux amd64:
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
Siapin project dan install SDK:
mkdir approval-agent && cd approval-agent
python3 -m venv .venv
source .venv/bin/activate
pip install temporalio openai
Nyalain dev server di terminal sendiri dan biarin tetap jalan:
temporal server start-dev
Server dengerin localhost:7233. Web UI ada di http://localhost:8233, itu tempat kamu ngeliat history workflow nanti.
Langkah 2: definisiin data yang nyebrang batas
Tiga dataclass. Bikin sesederhana mungkin dan pastikan serializable, karena setiap field bakal masuk 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 = ""
Langkah 3: taruh panggilan model di activity
Activity itu tempat side effect: HTTP call, write ke database, command shell. Temporal retry activity yang gagal, jadi activity harus idempotent.
Activity proposal manggil Responses API pakai JSON schema kalau ada key, dan fallback ke pengecekan keyword kalau nggak ada. Fallback-nya bukan mainan. Itu bikin kamu bisa ngetes alur approval tanpa bakar token, dan tutorial ini tetap bisa dijalanin tanpa internet.
# 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 itu tempat pesan Slack atau email masuk. Di sample ini dia nulis ke log worker, cukup buat ngeliat urutannya.
Langkah 4: tulis workflow-nya
Bagian ini yang selamat dari crash. Workflow yang pegang request ID, wait, dan keputusan.
# 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,
}
Ada empat detail yang penting di sini.
str(workflow.uuid4()) bikin ID tetap deterministik sekaligus jadi string. Pertama kali saya tulis tanpa cast, dan saya sempat lama debug approval yang diem-diem diabaikan. Ceritanya di bawah.
Signal handler ngecek request_id sebelum nyimpen apa pun. Klik telat dari halaman approval lama nggak bisa nge-approve aksi baru.
if self._decision is None bikin keputusan pertama jadi final. Klik dobel nggak bikin aksi jalan dua kali.
workflow.wait_condition dengan timeout ngubah diam jadi hasil yang jelas. Nggak ada approval berarti nggak ada eksekusi, dan result-nya bilang kenapa.
Langkah 5: worker dan starter
Worker daftarin workflow dan tiga activity ke satu 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())
Jalanin worker di satu terminal:
python worker.py
Langkah 6: jalanin
Di terminal kedua, mulai request yang jelas nggak boleh auto-approve:
python starter.py "Delete all test data from the production database" --timeout 600
workflow_id=approval-ff8267d6-7a1d-495f-b819-026c0107d234
Worker nyatet approval request, terus diem. Workflow-nya bukan polling. Dia parkir, nunggu event, dan nggak makan resource selama parkir.
Cek state dari client mana pun, ini yang bakal dilakuin UI:
temporal workflow query --workflow-id "$WF" --type status
Query result:
QueryResult {"request_id":"dda95a8b-b256-42b6-814b-6ace48b942ee","state":"awaiting_approval"}
Kirim approval pakai helper script:
python send_approval.py "$WF" "$RID" approve "checked the table name" --reviewer arya
Atau kirim dari apa pun yang bisa manggil CLI, cara ini yang dipake bot Slack atau job CI:
temporal workflow signal --workflow-id "$WF" --name approval_decision \
--input '{"request_id":"<request-id>","approved":true,"reviewer":"cli","reason":"on-call checked"}'
Ambil hasil akhirnya:
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"}
Coba request low risk, workflow langsung skip 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'}
Biar timer habis dan aksinya nggak pernah jalan:
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}
Langkah 7: matiin worker pas lagi nunggu
Ini alasan pakai Temporal instead of dict. Mulai request, pastiin statusnya waiting, terus matiin proses worker:
kill <worker-pid>
Workflow tetap terbuka. State-nya ada di history Temporal service, bukan di proses yang udah mati. Query tetap jawab:
QueryResult {"request_id":"e938f3ea-95ff-4f80-a1a2-619595dbbdec","state":"awaiting_approval"}
Nyalain worker baru, kirim approval, dan workflow selesai dengan request ID yang sama:
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"}
Nggak ada kode recovery yang perlu ditulis, dan nggak ada tabel pending approval yang harus dijaga sinkron. History-nya itu sendiri yang jadi state.
Langkah 8: bug yang bikin approval hilang
Versi pertama saya pakai baris ini:
self._request_id = workflow.uuid4()
Signal bawa string dari JSON. Workflow megang object uuid.UUID. != balikin true, jadi setiap keputusan dibuang, dan warning-nya ngeprint nilai yang sama dua kali:
ignoring decision for stale request 5b1b1df2-b092-4933-ae78-621bce2e1f9f (current 5b1b1df2-b092-4933-ae78-621bce2e1f9f)
Log-nya kelihatan kayak perbandingan rusak, karena memang rusak. workflow.uuid4() balikin uuid.UUID di temporalio. Cast ke str di batas masuk, atau parse payload signal jadi UUID sebelum dibandingin. Pilih satu dan konsisten buat semua ID yang masuk ke workflow.
Langkah 9: yang perlu di-hardening sebelum production
- Autentikasi pengirim signal. Signal itu cuma API call. Taruh endpoint approval di belakang auth layer dan jangan expose namespace Temporal.
- Pakai
request_idsebagai idempotency key buat tool asli. Kalau activity retry setelah timeout, tool harus kenal key yang sama. - Catat field keputusan di result, bukan cuma "approved". Reviewer, timestamp, dan alasan itu yang diminta audit.
- Simpan timeout di input, jangan hardcode. Aksi berbeda butuh kesabaran berbeda.
- Tambah query buat teks aksi yang pending, biar UI nampilin apa yang mau di-approve, bukan cuma ID.
Kapan cukup pakai tabel database
Kalau kamu cuma punya satu langkah approval, satu worker, dan Postgres yang udah jalan, tabel dengan kolom status, request_id, plus cron poller lebih simpel daripada nambah satu service. Temporal baru worth it kalau workflow-nya punya beberapa langkah, wait panjang, retry di sekitar tool yang labil, atau ada kebutuhan audit yang minta urutan kejadiannya bisa di-replay persis.
Interrupt plus checkpointer di LangGraph juga bisa ngelakuin hal serupa di dalam graph. Kalau agent-mu udah berupa graph LangGraph, pakai itu. Kalau approval gate-nya berdiri di antara service beda, job shell, dan panggilan LLM, Temporal naruh orkestrasinya di luar kode model, tempat yang lebih gampang dites.
Next step
Sambungin signal-nya ke tempat tim kamu udah nongkrong. Handler tombol Slack yang manggil handle.signal("approval_decision", ...) dengan request ID cuma sekitar tiga puluh baris. Tambah query buat teks proposal, dan pesannya bisa nampilin apa yang bakal jalan. Setelah itu beres, workflow yang sama bisa jalan di Temporal Cloud cuma dengan ganti alamat client dan credential, tanpa ubah kode.
Referensi
- Human-in-the-loop AI agent recipe: https://docs.temporal.io/ai/cookbook/human-in-the-loop-python
- Workflow message passing (signal, query, update): https://docs.temporal.io/develop/python/message-passing
- Setup Temporal Service lokal dengan 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