feat(docs): python support agent cookbook example (#3623)

* feat(python): add support_agent example with durable reply handling

- add support_agent durable workflow example
- add triage, reply generation, and escalation tasks
- use consider_events_since for early scoped reply events
- add trigger script for the example
- add E2E tests for resolved and timeout paths
- register support_agent workflows in the central example worker

* docs(cookbooks): add support agent workflow cookbook and example snippets

* docs(cookbooks): add support agent sidebar entry

* docs(cookbooks): minor improvement to support agent intro and fallback wording

* docs(cookbooks): document support agent worker registration

* docs(cookbooks): document support agent worker registration

* docs(lint): Fixed black errors

* docs(cookbooks): Improve phrasing and explanations

Also include how to run the tests themselves.

* docs(cookbooks): replace ASCII diagram with Mermaid

* docs(cookbooks): Generalize intro beyond support workflows

* docs(cookbooks): add links to support agent cookbook

* docs(cookbooks): clarify durable workflow rationale

Author: BloggerBust

examples/python/support_agent/trigger.py

@@ -0,0 +1,32 @@
+# > Trigger the workflow
+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 workflow
+ref = 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 complete
+result = ref.result()
+print(f"Workflow completed: {result}")
+
+

examples/python/support_agent/worker.py

@@ -0,0 +1,191 @@
+import os
+from datetime import timedelta
+from typing import Any
+
+from pydantic import BaseModel
+
+from hatchet_sdk import (
+    Context,
+    DurableContext,
+    Hatchet,
+    SleepCondition,
+    UserEventCondition,
+    or_,
+)
+
+hatchet = Hatchet()
+
+REPLY_EVENT_KEY = "support:customer-reply"
+TIMEOUT_SECONDS = 5
+
+
+# > Models
+class SupportTicketInput(BaseModel):
+    ticket_id: str
+    customer_email: str
+    subject: str
+    body: str
+
+
+class TriageOutput(BaseModel):
+    category: str
+    priority: str
+
+
+class ReplyOutput(BaseModel):
+    message: str
+
+
+class EscalationOutput(BaseModel):
+    reason: str
+    assigned_to: str
+
+
+
+
+# > Triage task
+@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)
+
+
+
+
+# > Generate reply task
+@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 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)
+
+
+
+
+# > Escalate task
+@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",
+    )
+
+
+
+
+# > Support agent workflow
+@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=5)
+
+    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,
+    }
+
+
+
+
+# > Worker registration
+def main() -> None:
+    worker = hatchet.worker(
+        "support-agent-worker",
+        workflows=[support_agent, triage_ticket, generate_reply, escalate_ticket],
+    )
+    worker.start()
+
+
+if __name__ == "__main__":
+    main()
+
+

examples/python/worker.py

@@ -76,6 +76,12 @@
 from examples.run_details.worker import run_detail_test_workflow
 from examples.serde.worker import serde_workflow
 from examples.simple.worker import simple, simple_durable
+from examples.support_agent.worker import (
+    escalate_ticket,
+    generate_reply,
+    support_agent,
+    triage_ticket,
+)
 from examples.timeout.worker import refresh_timeout_wf, timeout_wf
 from examples.webhook_with_scope.worker import (
     webhook_with_scope,
@@ -171,6 +177,10 @@ def main() -> None:
             otel_simple_task,
             otel_spawn_parent,
             otel_workflow,
+            support_agent,
+            triage_ticket,
+            generate_reply,
+            escalate_ticket,
         ],
         lifespan=lifespan,
     )

frontend/docs/pages/cookbooks/_meta.js

@@ -7,4 +7,9 @@ export default {
   "webhooks-stripe": "Stripe",
   "webhooks-github": "GitHub",
   "webhooks-slack": "Slack",
+  "--workflow-patterns": {
+    title: "Workflow Patterns",
+    type: "separator",
+  },
+  "workflow-support-agent": "Support Agent",
 };

frontend/docs/pages/cookbooks/index.mdx

@@ -23,3 +23,14 @@ Receive webhooks from external services and use them to trigger tasks in Hatchet
     Respond to Slack events like messages, reactions, and slash commands.
   </Cards.Card>
 </Cards>
+
+## Workflow Patterns
+
+Build practical end-to-end workflows with Hatchet’s durable execution, event handling, and task orchestration primitives.
+
+<Cards>
+  <Cards.Card title="Support Agent" href="/cookbooks/workflow-support-agent">
+    Build a durable support workflow that triages tickets, waits for customer
+    replies, and escalates on timeout.
+  </Cards.Card>
+</Cards>

frontend/docs/pages/cookbooks/workflow-support-agent.mdx

@@ -0,0 +1,115 @@
+import { Steps } from "nextra/components";
+import { snippets } from "@/lib/generated/snippets";
+import { Snippet } from "@/components/code";
+
+# 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:
+
+```mermaid
+flowchart TD
+    A[Support ticket received] --> B[Triage the ticket]
+    B --> C[Generate initial reply]
+    C --> D[Wait for reply or timeout]
+    D --> E[Customer reply]
+    D --> F[Timeout fires]
+    E --> G[Resolve ticket]
+    F --> H[Escalate to human support]
+```
+
+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
+
+<Steps>
+
+### Prepare your environment
+
+To run this example, you will need:
+
+- a working local Hatchet environment or access to [Hatchet Cloud](https://cloud.onhatchet.run)
+- the Python SDK example environment (see the [Quickstart](/v1/quickstart))
+- optionally, an `ANTHROPIC_API_KEY`
+
+Without `ANTHROPIC_API_KEY`, the example still runs using a fixed fallback reply.
+
+### Define the models
+
+Start with a few small Pydantic models for the workflow input and outputs.
+
+<Snippet src={snippets.python.support_agent.worker.models} />
+
+The 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](/v1/tasks).
+
+First, add a task to classify the incoming ticket:
+
+<Snippet src={snippets.python.support_agent.worker.triage_task} />
+
+Next, add a task to generate the initial support reply. This task calls Claude when `ANTHROPIC_API_KEY` is present and falls back to a fixed response when it is not.
+
+<Snippet src={snippets.python.support_agent.worker.generate_reply_task} />
+
+Finally, add a task to represent escalation to the support team:
+
+<Snippet src={snippets.python.support_agent.worker.escalate_task} />
+
+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](/v1/durable-execution). 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.
+
+<Snippet src={snippets.python.support_agent.worker.support_agent_workflow} />
+
+As you can see, 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 [`consider_events_since`](/v1/durable-event-waits#lookback-windows) on the reply event condition. A customer reply could arrive while the workflow is still finishing triage or generating the first response. By using `consider_events_since`, 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.
+
+<Snippet src={snippets.python.support_agent.worker.worker_registration} />
+
+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.
+
+<Snippet src={snippets.python.support_agent.trigger.trigger_the_workflow} />
+
+Because the workflow uses `consider_events_since`, 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 Python SDK examples locally, you can run the support agent tests with:
+
+```bash
+pytest examples/support_agent/test_support_agent.py
+```
+
+Together, these tests validate both branches of the workflow and confirm that early reply events are handled safely without coordination sleeps.
+
+</Steps>
+
+## 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.