How to Create a Support Agent Using Hatchet
Many real-world workflows become difficult to manage once they involve multiple steps, long waits, human replies, and escalation rules. Support is one example, but the same pattern also shows up in onboarding, approvals, incident response, and other operational flows. In this cookbook, we will build a simple support agent that triages a ticket, generates an initial reply, and then waits for either a customer response or a timeout. If the customer replies, the workflow resolves. If no reply arrives in time, the workflow escalates the ticket to a human support agent.
What this example builds
This example implements the following durable support workflow:
Hatchet's durable execution model helps keep the whole interaction in one workflow rather than scattering it across separate queue jobs and ad hoc timers.
Setup
Prepare your environment
To run this example, you will need:
- a working local Hatchet environment or access to Hatchet Cloud
- a Hatchet SDK example environment (see the Quickstart)
- optionally, an
ANTHROPIC_API_KEYfor live LLM replies
Without ANTHROPIC_API_KEY, the example runs using a fixed fallback reply. To use the live Claude path, you also need the Anthropic SDK installed for your language.
Define the models
Start by defining the types for the workflow input and task outputs.
class SupportTicketInput(BaseModel): ticket_id: str customer_email: str subject: str body: strclass TriageOutput(BaseModel): category: str priority: strclass ReplyOutput(BaseModel): message: strclass EscalationOutput(BaseModel): reason: str assigned_to: strThe models keep the inputs and outputs for each task explicit, which makes the workflow easier to inspect and test.
Add the workflow tasks
The durable workflow delegates its work to a few small tasks.
First, add a task to classify the incoming ticket:
@hatchet.task(input_validator=SupportTicketInput)async def triage_ticket(input: SupportTicketInput, ctx: Context) -> TriageOutput: """Classify the ticket into a category and priority.""" subject = input.subject.lower() body = input.body.lower() text = subject + " " + body if any(word in text for word in ["bill", "charge", "payment", "invoice"]): category = "billing" elif any(word in text for word in ["login", "password", "auth", "access"]): category = "account" else: category = "technical" if any(word in text for word in ["urgent", "critical", "down", "outage"]): priority = "high" elif any(word in text for word in ["twice", "broken", "error"]): priority = "medium" else: priority = "low" return TriageOutput(category=category, priority=priority)Next, add a task to generate the initial support reply. When ANTHROPIC_API_KEY is set, the task calls Claude to produce the reply. Otherwise it returns a fixed fallback response.
@hatchet.task(input_validator=SupportTicketInput)async def generate_reply(input: SupportTicketInput, ctx: Context) -> ReplyOutput: """Generate an initial support reply using Claude.""" api_key = os.environ.get("ANTHROPIC_API_KEY") if not api_key: return ReplyOutput( message=f"Thank you for contacting support about: {input.subject}. " "We are looking into this and will get back to you shortly." ) import importlib anthropic = importlib.import_module("anthropic") client = anthropic.AsyncAnthropic(api_key=api_key) response = await client.messages.create( model="claude-sonnet-4-20250514", max_tokens=300, messages=[ { "role": "user", "content": ( f"You are a friendly support agent. Write a brief, helpful initial " f"reply to this support ticket.\n\n" f"Subject: {input.subject}\n" f"Message: {input.body}\n\n" f"Keep the reply under 3 sentences." ), } ], ) text = response.content[0].text return ReplyOutput(message=text)Finally, add a task to represent escalation to the support team:
@hatchet.task(input_validator=SupportTicketInput)async def escalate_ticket(input: SupportTicketInput, ctx: Context) -> EscalationOutput: """Escalate an unresolved ticket to the human support team.""" return EscalationOutput( reason=f"No customer reply within {TIMEOUT_SECONDS}s timeout", assigned_to="support-team@example.com", )Keeping triage, reply generation, and escalation as separate tasks keeps the workflow itself small and makes each piece easier to reason about.
Build the durable workflow
Now tie everything together in a durable Hatchet workflow. A durable workflow is a good fit here because this interaction may stay open for some time while waiting for a customer reply. Hatchet persists the workflow state and its wait conditions, so the workflow can survive long delays, worker restarts, or even a worker crash, then continue later on another worker. That gives you a straightforward way to model the whole interaction without adding custom recovery logic.
@hatchet.durable_task(input_validator=SupportTicketInput)async def support_agent( input: SupportTicketInput, ctx: DurableContext) -> dict[str, Any]: # Step 1: Triage the ticket triage = await triage_ticket.aio_run(input) # Step 2: Generate an initial reply reply = await generate_reply.aio_run(input) # Step 3: Wait for a customer reply or timeout now = await ctx.aio_now() consider_events_since = now - timedelta(minutes=LOOKBACK_MINUTES) wait_result = await ctx.aio_wait_for( "await-customer-reply", or_( SleepCondition(timedelta(seconds=TIMEOUT_SECONDS)), UserEventCondition( event_key=REPLY_EVENT_KEY, scope=input.ticket_id, consider_events_since=consider_events_since, ), ), ) # The or-group result is {"CREATE": {"<condition_key>": ...}}. # Check whether the reply event condition was the one that resolved. resolved_key = list(wait_result["CREATE"].keys())[0] customer_replied = resolved_key == REPLY_EVENT_KEY if not customer_replied: # Step 4a: Timeout -> escalate await escalate_ticket.aio_run(input) return { "ticket_id": input.ticket_id, "status": "escalated", "triage_category": triage.category, "triage_priority": triage.priority, "initial_reply": reply.message, } # Step 4b: Customer replied -> resolve return { "ticket_id": input.ticket_id, "status": "resolved", "triage_category": triage.category, "triage_priority": triage.priority, "initial_reply": reply.message, }The workflow runs triage first, generates an initial reply, and then waits for one of two things to happen: either a customer reply event arrives for that ticket, or the timeout fires. From there, the workflow either resolves the ticket or escalates it.
The detail that matters most here is the lookback window on the reply event condition. A customer reply could arrive while the workflow is still finishing triage or generating the first response. By using a lookback window (consider_events_since in Python, considerEventsSince in TypeScript), the workflow can still pick up that reply once the wait becomes active instead of missing it because the event arrived slightly early.
Register and start the worker
To run this workflow, register the workflow and its tasks on a Hatchet worker, then start it.
def main() -> None: worker = hatchet.worker( "support-agent-worker", workflows=[support_agent, triage_ticket, generate_reply, escalate_ticket], ) worker.start()if __name__ == "__main__": main()In TypeScript, workflows are registered through the shared example worker rather than a per-example registration file.
With the worker running, you can trigger the workflow and observe either the resolved or escalated outcome.
Trigger the workflow
The example also includes a small trigger script that starts the workflow, pushes a scoped reply event, and waits for the result.
from examples.support_agent.worker import ( REPLY_EVENT_KEY, SupportTicketInput, hatchet, support_agent,)ticket = SupportTicketInput( ticket_id="ticket-42", customer_email="alice@example.com", subject="Login broken", body="I can't log in since this morning.",)# Start the support agent workflowref = support_agent.run(ticket, wait_for_result=False)print(f"Started workflow run: {ref.workflow_run_id}")# Push a customer reply event (scoped to this ticket)print("Pushing customer reply event...")hatchet.event.push( REPLY_EVENT_KEY, {"message": "I cleared my cookies and it works now. Thanks!"}, scope=ticket.ticket_id,)# Wait for the workflow to completeresult = ref.result()print(f"Workflow completed: {result}")Because the workflow uses a lookback window, the trigger can push the reply event immediately after starting the support agent.
Test it
This example includes two end-to-end tests against a live Hatchet instance:
- a resolved path, where the customer reply event arrives before the timeout
- a timeout path, where no reply arrives and the workflow escalates
If you are running the SDK examples locally:
pytest examples/support_agent/test_support_agent.pyTogether, these tests validate both branches of the workflow and confirm that early reply events are handled safely without coordination sleeps.
Why Hatchet fits this workflow
The interesting part of this example is not the LLM call. It is the combination of waiting, branching, and keeping the full interaction in one place. A support flow like this usually needs to preserve state across several steps, wait for human input, and react differently depending on whether a reply arrives before a deadline. Hatchet fits that pattern well because you can express the event wait and timeout branch directly in the workflow. That makes the control flow easier to inspect, easier to test, and easier to extend as the interaction becomes more complex.
Next steps
A natural next step would be to connect this workflow to a real ticketing system and carry the conversation beyond a single reply. You could also make escalation depend on the content of the customer response instead of only on timeout. For this cookbook, though, the smaller version is enough to show the core pattern: start work immediately, wait safely for a reply, and escalate when the deadline passes.
Last updated on August 13, 2026