Building autonomous agents that handle simple, single-turn prompts is easy. The real headache starts when your agent has to wait three days for a human manager to approve an action, or ten minutes for an external API to process a batch of files. If your agent holds its context in memory while waiting, a single server restart wipes out everything. I learned this the hard way when a deployment killed dozens of active workflows mid-execution, leaving users stranded with half-finished tasks.
To build reliable long running agents, you need a system that decouples execution from memory. Your agent must be able to pause, write its full context to storage, yield control, and wake up only when a trigger or callback arrives.
Here is how you can implement durable agent state management and asynchronous callbacks step by step.
Traditional agent scripts run in a continuous loop: think, act, observe, repeat. This works fine for quick scripts, but fails for long-running processes.
[ Traditional Agent Loop ]
Agent Start -> Prompt LLM -> Call Tool -> Wait for API (Blocking) -> Process -> Done
(If process crashes during Wait, state is lost forever)
[ Event-Driven Durable Agent ]
Agent Start -> Prompt LLM -> Save State to DB -> Suspend Process
|
Webhook / Human Input arrives later -----------------+
v
Resume Process -> Load State -> Complete Task
To handle asynchronous callbacks, we break this single process into three main parts:
- State Persistence: Saving the memory, scratchpad, and conversation history to a database.
- Execution Suspension: Halting execution cleanly without holding open HTTP connections or thread locks.
- Event Rehydration: Rebuilding the agent's full state when a webhook or user input triggers a wake-up call.
If you are scaling complex AI agent development, getting this foundation right early saves hundreds of hours of debugging downstream.
Step 1: Define a Serializable Agent State Schema
You cannot save raw LLM instances or active client connections to a database. You must serialize the agent state into a structured schema containing the prompt history, tool call outputs, pending callbacks, and current status.
Here is a Pydantic schema using Python that captures what a durable execution setup needs:
from enum import Enum
from typing import Dict, Any, List, Optional
from pydantic import BaseModel, Field
import datetime
class AgentStatus(str, Enum):
RUNNING = "running"
SUSPENDED = "suspended"
COMPLETED = "completed"
FAILED = "failed"
class AgentState(BaseModel):
thread_id: str
status: AgentStatus = AgentStatus.RUNNING
system_prompt: str
messages: List[Dict[str, Any]] = Field(default_factory=list)
pending_callback_id: Optional[str] = None
context_data: Dict[str, Any] = Field(default_factory=dict)
created_at: str = Field(default_factory=lambda: datetime.datetime.utcnow().isoformat())
updated_at: str = Field(default_factory=lambda: datetime.datetime.utcnow().isoformat())
This model stores everything necessary to rebuild the agent environment. The pending_callback_id field tracks external async actions like email responses, third-party webhook processing, or human approvals.
Step 2: Implement Persistent Storage for Agent State Management
Next, build a persistence repository. SQLite works great for local testing, while PostgreSQL or Redis fits production workloads.
This simple repository pattern manages durable state transitions:
import sqlite3
import json
class StateRepository:
def __init__(self, db_path: str = "agent_state.db"):
self.db_path = db_path
self._init_db()
def _init_db(self):
with sqlite3.connect(self.db_path) as conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS states (
thread_id TEXT PRIMARY KEY,
status TEXT,
pending_callback_id TEXT,
payload TEXT
)
""")
conn.commit()
def save(self, state: AgentState):
state.updated_at = datetime.datetime.utcnow().isoformat()
with sqlite3.connect(self.db_path) as conn:
conn.execute(
"""
INSERT INTO states (thread_id, status, pending_callback_id, payload)
VALUES (?, ?, ?, ?)
ON CONFLICT(thread_id) DO UPDATE SET
status=excluded.status,
pending_callback_id=excluded.pending_callback_id,
payload=excluded.payload
""",
(
state.thread_id,
state.status.value,
state.pending_callback_id,
json.dumps(state.dict())
)
)
conn.commit()
def load(self, thread_id: str) -> Optional[AgentState]:
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
cursor.execute("SELECT payload FROM states WHERE thread_id = ?", (thread_id,))
row = cursor.fetchone()
if row:
return AgentState(**json.loads(row[0]))
return None
With persistent storage ready, the agent can stop execution entirely while waiting for asynchronous events.
Step 3: Handle Asynchronous Callbacks and Suspensions
When your agent hits an async barrier, such as waiting for a human review or calling a long-running batch extraction API, it should emit an external request, register a unique callback key, save state, and exit.
Here is an example showing how an agent suspends its flow:
import uuid
class AgentEngine:
def __init__(self, repo: StateRepository):
self.repo = repo
def execute_human_approval_step(self, thread_id: str, action_details: str):
state = self.repo.load(thread_id)
if not state:
raise ValueError("Thread not found")
# Create a unique callback listener token
callback_id = f"cb_{uuid.uuid4().hex[:8]}"
# Record action intent into conversation history
state.messages.append({
"role": "assistant",
"content": f"Requesting human approval for action: {action_details}"
})
# Transition state to suspended
state.status = AgentStatus.SUSPENDED
state.pending_callback_id = callback_id
# Save state to storage
self.repo.save(state)
print(f"[Agent Suspended] Task waiting for callback token: {callback_id}")
return callback_id
Instead of blocking thread execution, this design frees memory and CPU resources. It is especially useful when integrating complex backend tasks with workflow automation pipelines that spans across external services.
Step 4: Resume Execution on Callback Arrival
Now, set up a handler to accept incoming callbacks, find the matching agent execution state, inject the incoming payload, and resume processing.
def handle_callback(self, callback_id: str, payload: Dict[str, Any]):
# Search state database for matching callback token
with sqlite3.connect(self.repo.db_path) as conn:
cursor = conn.cursor()
cursor.execute("SELECT thread_id FROM states WHERE pending_callback_id = ?", (callback_id,))
row = cursor.fetchone()
if not row:
raise ValueError(f"No agent suspended for callback ID: {callback_id}")
thread_id = row[0]
state = self.repo.load(thread_id)
# Update thread state with incoming payload
state.messages.append({
"role": "user",
"content": f"Callback received with data: {json.dumps(payload)}"
})
# Clear callback lock and set back to running
state.pending_callback_id = None
state.status = AgentStatus.RUNNING
self.repo.save(state)
print(f"[Agent Resumed] Processing thread: {thread_id}")
return self.continue_agent_loop(state)
def continue_agent_loop(self, state: AgentState):
# Continue reasoning loop with updated state context
state.messages.append({
"role": "assistant",
"content": "Received callback approval. Proceeding with execution."
})
state.status = AgentStatus.COMPLETED
self.repo.save(state)
return state
Step 5: Put It Together and Run an Simulation
Here is a full demonstration of the cycle: starting an agent task, pausing for external input, and resuming execution.
if __name__ == "__main__":
repo = StateRepository()
engine = AgentEngine(repo)
# Initialize a new agent run
thread_id = f"session_{uuid.uuid4().hex[:6]}"
initial_state = AgentState(
thread_id=thread_id,
system_prompt="You are an autonomous assistant handling invoice operations.",
messages=[{"role": "user", "content": "Process refund for Order #9482"}]
)
repo.save(initial_state)
# Agent runs until it needs external authorization
callback_token = engine.execute_human_approval_step(
thread_id=thread_id,
action_details="Refund $450 to customer"
)
# At this point, the application process can close, restart, or crash.
# The agent state remains safe inside SQLite.
# Later, an incoming webhook or API call fires this endpoint handler:
print("\n... Waiting for external webhook callback ...\n")
incoming_webhook_payload = {
"approved": True,
"approver": "manager@company.com"
}
final_state = engine.handle_callback(callback_token, incoming_webhook_payload)
print(f"Final Agent Status: {final_state.status.value}")
Running this code prints:
[Agent Suspended] Task waiting for callback token: cb_a1b2c3d4
... Waiting for external webhook callback ...
[Agent Resumed] Processing thread: session_e5f6g7
Final Agent Status: completed
Best Practices for Production Durable Execution
When building production grade systems for async agent workflows, keep these tips in mind:
- Use Dedicated Task Orchestrators: While SQLite works for basic usage, engines like Temporal, Prefect, or AWS Step Functions handle re-runs, retries, and time-outs much better.
- Store Minimal Payload Context: Do not save large binary outputs directly inside the agent execution payload. Store them in S3 or object storage, then pass references into the agent state object.
- Handle Timeouts: Always append expiration timers to suspended states. If a callback never returns, a fallback background job should wake the agent up to handle the error condition gracefully.
Building long running agents requires shifting from simple synchronous loops to robust, stateful architectures. If your team needs hand building reliable agent infrastructures, our engineers at Gaper can help design state management patterns suited for enterprise scale.
Top comments (0)