Customer support automation

A durable customer-support pipeline — email ingest, agent triage, draft response, human approval, send. Built around Flux's pause/resume primitive for the human step.

We will build a customer-support pipeline where an inbound email is triaged by an LLM, a draft response is generated, a human operator approves or edits the draft, and the workflow sends the final reply.

The interesting bit is the human step. In a queue-based system, you would have to invent your own state machine, schedule a callback, store the draft somewhere, and reconcile what came back. In Flux, you call pause() and let the workflow sit. When the operator resumes it with their decision, execution picks up on the next line.

What you are building

A workflow per inbound email. The workflow runs in five stages: ingest, triage, draft, pause for human approval, send. The pause is durable — it can sit for hours or days while the operator is on vacation, and nothing leaks. The triage and draft stages use the Flux agent() primitive against your provider of choice.

email -> triage -> draft -> [PAUSE: operator] -> send

  1. Ingest

    The ingest task pulls a single message off IMAP and returns it as a structured dict. We retry on transient IMAP failures and time out aggressively so that one stuck connection does not block the worker.

    import imaplib
import email
from email.message import EmailMessage
from flux.task import task


@task.with_options(retry_max_attempts=3, retry_delay=2, timeout=30)
async def fetch_email(uid: str) -> dict:
    with imaplib.IMAP4_SSL("imap.example.com") as conn:
        conn.login("support@example.com", "{password}")
        conn.select("INBOX")
        _, data = conn.fetch(uid, "(RFC822)")
        raw = data[0][1]

    msg = email.message_from_bytes(raw, _class=EmailMessage)
    body = msg.get_body(preferencelist=("plain",))
    return {
        "uid": uid,
        "from": msg["From"],
        "subject": msg["Subject"],
        "body": body.get_content() if body else "",
    }

    In production you would have a separate worker process polling the inbox, posting a new workflow execution per unread UID. The polling loop is not part of the workflow — it is a service that creates work for the workflow to do. We will come back to that in the operational notes at the end.

    Treating fetch as an explicit task (instead of pulling the email body inside the workflow function directly) means the IMAP call is replayed from the event log on a retry, not re-issued against the server. That matters when the email already moved out of the INBOX between executions, or when a transient network blip would otherwise trigger a re-fetch of a message the workflow has already processed.

  2. Triage

    Triage classifies the email into one of five categories. Each category gets a different downstream tone, urgency, and escalation policy.

    from pydantic import BaseModel
from flux.tasks.ai import agent


class Triage(BaseModel):
    category: str  # "complaint" | "refund" | "technical" | "sales" | "spam"
    urgency: str   # "low" | "medium" | "high"
    summary: str


async def make_triage_agent():
    return await agent(
        system_prompt=(
            "You triage inbound customer emails. "
            "Choose the single best category and assign an urgency. "
            "Write a one-sentence summary of what the customer wants."
        ),
        model="anthropic/claude-sonnet-4-20250514",
        response_format=Triage,
        max_tool_calls=1,
        max_tokens=512,
        stream=False,
    )

    max_tool_calls=1 caps the agent loop. Triage does not have tools and does not need a multi-turn conversation; capping it keeps a triage call from turning into a multi-turn tool-use exchange.

    The Anthropic provider enforces response_format at the API by issuing a forced tool call, so the model returns JSON matching the schema. max_tokens=512 is honored as well. For triage that is the right shape — a structured Triage object comes back ready to use.

  3. Draft response

    A second agent call generates the draft, with the triage result as context so the tone matches.

    async def make_draft_agent():
    return await agent(
        system_prompt=(
            "You draft customer-support replies. "
            "Match the tone implied by the triage category — empathetic for "
            "complaints, factual for technical, concise for sales. "
            "Never promise refunds, discounts, or timelines without operator confirmation."
        ),
        model="anthropic/claude-sonnet-4-20250514",
        max_tokens=1024,
        stream=False,
    )

    The “never promise” guardrail is non-negotiable. The draft is going to a human reviewer anyway, but you want the LLM to produce something they can rubber-stamp in the easy 70% of cases rather than something they have to rewrite from scratch. Push the policy into the system prompt and the draft quality goes up sharply.

  4. Human approval gate

    This is the load-bearing step. We pause the workflow and wait for an operator to resume it.

    from flux.tasks import pause


# inside the workflow:
decision = await pause(
    name="operator_approval",
    output={
        "triage": triage.model_dump(),
        "draft": draft,
        "customer_email": email_data,
    },
)

    pause(name, output) does two things. It raises PauseRequested to suspend the workflow, and it attaches output to the pause event so the operator UI (or anything reading the event log) has full context to display. The workflow goes into PAUSED state and stays there indefinitely.

    Resuming is a separate API call. The CLI takes three positional arguments — the workflow name, the execution ID, and the resume input: flux workflow resume handle_support_email <execution_id> '{...}'. Whatever JSON you pass as the resume input becomes the return value of the pause() call. So if the operator UI sends back {"approved": true, "modifications": null}, the decision variable in the workflow gets exactly that dict.

    decision = {
    "approved": True,
    "modifications": None,  # or a rewritten draft string
}

    There is no timeout on the pause itself. The workflow can sit paused for an hour, a day, a week. The worker is not holding any in-memory state — the workflow context is fully on disk. When the resume arrives, a worker claims it and replays from the pause.

  5. Send

    The send step actually transmits the reply. The draft might have been modified by the operator, so the workflow reads from the resume input before sending.

    import smtplib
from email.message import EmailMessage


@task.with_options(retry_max_attempts=3, retry_delay=5, timeout=30)
async def send_reply(to: str, subject: str, body: str) -> dict:
    msg = EmailMessage()
    msg["To"] = to
    msg["From"] = "support@example.com"
    msg["Subject"] = f"Re: {subject}"
    msg.set_content(body)

    with smtplib.SMTP_SSL("smtp.example.com") as server:
        server.login("support@example.com", "{password}")
        server.send_message(msg)

    return {"sent": True, "to": to, "subject": msg["Subject"]}

    Sending is the one truly irreversible step. We retry on transient SMTP failures, but we do not roll back — once an email has gone out, there is no delete_email_row to call. The retry policy and the explicit human approval upstream do the work that rollback does for reversible side effects.

    If you wanted the LLM itself to be able to invoke send_reply (in a more agentic variant of this pipeline), you would declare the task with requires_approval=True:

    @task.with_options(requires_approval=True)
async def send_reply(...) -> dict: ...

    The engine emits a TASK_AWAITING_APPROVAL event before the body runs and pauses the workflow until an operator decides via flux execution approve|reject or the equivalent HTTP endpoint. For the workflow shown here we do not need it — the workflow-level pause() already gates the send.

    The two patterns — workflow-level pause and tool-level approval — solve different problems. Workflow-level pause is right when the human is reviewing a finished artifact (the draft) and approving the whole action. Tool-level approval is right when an autonomous agent might choose to invoke a sensitive tool mid-loop, and you want a human gate at the tool invocation rather than at the end of the run. Mix them when the workflow gives the agent freedom inside a bounded budget and the human still wants final say on irreversible actions.

  6. The workflow

    Everything wired together:

    from flux import ExecutionContext
from flux.workflow import workflow


@workflow
async def handle_support_email(ctx: ExecutionContext[dict]):
    params = ctx.input or {}
    uid = params["uid"]

    email_data = await fetch_email(uid)

    triage_agent = await make_triage_agent()
    triage = await triage_agent(
        instruction="Triage this email.",
        context=email_data["body"],
    )

    draft_agent = await make_draft_agent()
    draft = await draft_agent(
        instruction=(
            f"Draft a reply to this {triage.category} email. "
            f"Urgency: {triage.urgency}. Summary: {triage.summary}."
        ),
        context=email_data["body"],
    )

    decision = await pause(
        name="operator_approval",
        output={
            "triage": triage.model_dump(),
            "draft": draft,
            "customer_email": email_data,
        },
    )

    if not decision.get("approved"):
        return {
            "status": "rejected",
            "reason": decision.get("reason"),
            "uid": uid,
        }

    final_body = decision.get("modifications") or draft

    send_result = await send_reply(
        to=email_data["from"],
        subject=email_data["subject"],
        body=final_body,
    )

    return {
        "status": "sent",
        "uid": uid,
        "category": triage.category,
        **send_result,
    }

    The workflow is fifty lines. Most of the value is in what is not there: no state machine, no scheduling glue, no manual checkpointing, no reconciliation between the paused state and what the operator saw. Flux’s event log carries all of that.

  7. Operational notes

    A few things worth knowing before you run this in anger.

    The IMAP poll is not part of the workflow. You want a separate small worker process (or a cron-driven script) that polls the inbox every few minutes and starts a new handle_support_email execution per unread UID. Keep it stateless: the workflow’s own idempotency on uid handles dedup, and the inbox itself tracks read state.

    Use label-based worker affinity for the polling worker. Tag the poll worker with a label like role=poller and tag the main workers with role=processor. Then set affinity={"role": "processor"} on handle_support_email so polling and processing do not compete for the same worker slot. See worker affinity for the full pattern.

    Long pauses are fine. A worker is not holding the paused workflow in memory. The pause sits in the event log. A separate worker (potentially in a future deployment) picks up the resume when it arrives.

    Auto-triage some categories if volume demands it. For a high-volume inbox, you may want spam and certain low-risk categories to auto-resolve without the human gate. Branch on triage.category before the pause and route the auto-paths around it. The shape of the workflow stays the same; you are just choosing where the pause() call lives.

What to remember

pause() is a first-class workflow primitive, not a hack. The workflow goes into PAUSED state, releases the worker, and waits indefinitely. There is no timeout, no in-memory state, and no need to invent your own scheduling glue.

Agents are tasks. The result of agent(...) is a @task-decorated function. It records events, supports retries, fits inside parallel(...) and pipeline(...), and replays from cache like any other task.

Human-in-the-loop is a workflow design decision, not a feature. You decide where the human gate goes by where you put pause(). The runtime treats it as just another suspension point.

Where this shows up next