Batch Ticket Labeling with Claude and Postgres in Kestra

id: anthropic-message-batch-postgres-enrichment
namespace: company.team
description: |
  Labels pending support tickets in Postgres with an Anthropic Message Batch.
  Each run claims tickets, submits one request per ticket, waits for the batch to end, loads the results and writes a category back to each ticket.

inputs:
  - id: model
    type: STRING
    displayName: Model
    description: Claude model used for every request in the batch.
    defaults: "claude-haiku-5-5"

  - id: max_batch_size
    type: INT
    displayName: Max batch size
    description: Maximum number of tickets claimed and sent in one batch.
    defaults: 10000

  - id: max_batch_bytes
    type: INT
    displayName: Max batch bytes
    description: The most bytes of request JSON one run may build, kept well under
      Kestra's 1 MiB queue message limit to leave room for escaping.
    defaults: 600000
    min: 50000

  - id: approval_threshold
    type: INT
    displayName: Approval threshold
    description: When more tickets than this are waiting, the run pauses for manual
      approval unless a run was approved in the last 24 hours. Resuming approves
      large backlogs for 24 hours; a pause nobody resumes cancels the run after
      24 hours.
    defaults: 1000

  - id: max_attempts
    type: INT
    displayName: Max attempts
    description: Failed results (any type other than succeeded) allowed per ticket
      before it is marked failed instead of pending.
    defaults: 3

variables:
  labels: "billing,bug,feature_request,account,other"

tasks:
  - id: size_check
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: Counts claimable tickets, up to max_batch_size.
    url: "{{ secret('POSTGRES_URL') }}"
    username: "{{ secret('POSTGRES_USER') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    sql: |
      SELECT count(*) AS claimable
      FROM (
        SELECT 1 FROM support_tickets
        WHERE enrichment_status = 'pending'
           OR (enrichment_status = 'submitted' AND claimed_at < now() - interval '25 hours')
        LIMIT CAST(:max_batch_size AS integer)
      ) AS capped
    parameters:
      max_batch_size: "{{ inputs.max_batch_size }}"
    fetchType: FETCH_ONE

  - id: if_empty
    type: io.kestra.plugin.core.flow.If
    description: Ends the run when no ticket is waiting.
    condition: "{{ outputs.size_check.row.claimable == 0 }}"
    then:
      - id: nothing_to_claim
        type: io.kestra.plugin.core.execution.Exit
        state: SUCCESS

  - id: if_large_batch
    type: io.kestra.plugin.core.flow.If
    description: Asks for approval before a large backlog is sent, unless a run was
      approved in the last 24 hours.
    condition: "{{ outputs.size_check.row.claimable > inputs.approval_threshold and
      kv(flow.id ~ '-approval', namespace=flow.namespace, errorOnMissing=false)
      is null }}"
    then:
      - id: approve_batch
        type: io.kestra.plugin.core.flow.Pause
        description: Resume to submit the batch and approve large backlogs for 24 hours.
          Without approval within 24 hours the execution is canceled.
        pauseDuration: PT24H
        behavior: CANCEL

      - id: record_approval
        type: io.kestra.plugin.core.kv.Set
        description: While this key exists, large backlogs run without pausing.
        key: "{{ flow.id }}-approval"
        value: "{{ execution.id }}"
        ttl: PT24H

  - id: claim
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: |
      Marks claimable tickets as submitted, oldest first, while both max_batch_size and max_batch_bytes allow, and returns one MessageBatch request per ticket with its subject cut to 500 and its body to 4000 characters.
      Claimable means pending, or submitted with claimed_at older than 25 hours: a run stops waiting for its batch after 24 hours 30 minutes, so an older claim belongs to a run that died.
      Postgres does not allow FOR UPDATE with a window function, so the running byte total is computed in its own CTE. It counts 2 more bytes per request for the separator json_agg adds, so request_bytes never exceeds max_batch_bytes.
    url: "{{ secret('POSTGRES_URL') }}"
    username: "{{ secret('POSTGRES_USER') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    sql: |
      WITH claimable AS (
        SELECT id, created_at,
               json_build_object(
                 'customId', 'ticket-' || id,
                 'model', CAST(:model AS text),
                 'maxTokens', 1024,
                 'system', CAST(:prompt AS text),
                 'messages', json_build_array(json_build_object(
                   'type', 'USER',
                   'content', 'Subject: ' || left(subject, 500) || E'\n\n' || left(body, 4000)))
               ) AS request
        FROM support_tickets
        WHERE enrichment_status = 'pending'
           OR (enrichment_status = 'submitted' AND claimed_at < now() - interval '25 hours')
        ORDER BY created_at, id
        LIMIT CAST(:max_batch_size AS integer)
        FOR UPDATE SKIP LOCKED
      ), sized AS (
        SELECT id, request,
               sum(octet_length(CAST(request AS text)) + 2) OVER (ORDER BY created_at, id) AS running_bytes
        FROM claimable
      ), claimed AS (
        UPDATE support_tickets AS t
        SET enrichment_status = 'submitted',
            claimed_at = now(),
            enrichment_execution_id = :execution_id,
            enrichment_batch_id = NULL
        FROM sized AS s
        WHERE t.id = s.id
          AND s.running_bytes <= CAST(:max_batch_bytes AS bigint)
        RETURNING t.id, s.request
      )
      SELECT claimed, coalesce(octet_length(requests), 0) AS request_bytes, requests
      FROM (
        SELECT count(*) AS claimed, CAST(json_agg(request ORDER BY id) AS text) AS requests
        FROM claimed
      ) AS agg
    parameters:
      max_batch_size: "{{ inputs.max_batch_size }}"
      max_batch_bytes: "{{ inputs.max_batch_bytes }}"
      execution_id: "{{ execution.id }}"
      model: "{{ inputs.model }}"
      prompt: "You label customer support tickets. Reply with exactly one label from
        this list and nothing else: {{ vars.labels | replace({',': ', '}) }}"
    fetchType: FETCH_ONE

  - id: if_none_claimed
    type: io.kestra.plugin.core.flow.If
    description: Claim returns no tickets when another run took them after size_check.
    condition: "{{ outputs.claim.row.claimed == 0 }}"
    then:
      - id: none_claimed
        type: io.kestra.plugin.core.execution.Exit
        state: SUCCESS

  - id: create_batch
    type: io.kestra.plugin.anthropic.MessageBatch
    description: Submits one Message Batch with a request per claimed ticket.
    apiKey: "{{ secret('ANTHROPIC_API_KEY') }}"
    mode: CREATE
    requests: "{{ outputs.claim.row.requests }}"

  - id: record_batch
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: Stores the batch id on this run's claimed tickets.
    url: "{{ secret('POSTGRES_URL') }}"
    username: "{{ secret('POSTGRES_USER') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    sql: |
      UPDATE support_tickets
      SET enrichment_batch_id = :batch_id
      WHERE enrichment_execution_id = :execution_id
        AND enrichment_status = 'submitted'
    parameters:
      batch_id: "{{ outputs.create_batch.batchId }}"
      execution_id: "{{ execution.id }}"
    fetchType: NONE

  - id: wait_for_batch
    type: io.kestra.plugin.core.flow.LoopUntil
    description: Checks the batch every minute. A batch ends at most 24 hours after
      creation; maxDuration adds 30 minutes because once it has passed, Kestra
      fails the loop without checking the condition.
    condition: "{{ outputs.retrieve_batch.status == 'ended' }}"
    checkFrequency:
      interval: PT1M
      maxDuration: PT24H30M
    failOnMaxReached: true
    tasks:
      - id: retrieve_batch
        type: io.kestra.plugin.anthropic.MessageBatch
        description: Reads the batch status and, once the batch has ended, stores every
          result in an ION file.
        apiKey: "{{ secret('ANTHROPIC_API_KEY') }}"
        mode: RETRIEVE
        batchId: "{{ outputs.create_batch.batchId }}"
        fetchType: STORE
        retry:
          type: constant
          interval: PT1M
          maxAttempts: 3

  - id: load_results
    type: io.kestra.plugin.jdbc.postgresql.Batch
    description: batch_id is read from the ticket named in customId, so no value is
      spliced into the SQL.
    url: "{{ secret('POSTGRES_URL') }}"
    username: "{{ secret('POSTGRES_USER') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    from: "{{ outputs.retrieve_batch.uri }}"
    sql: |
      INSERT INTO ticket_enrichment_results (batch_id, custom_id, result_type, output_text, error_message)
      VALUES (
        (SELECT enrichment_batch_id FROM support_tickets WHERE id = CAST(split_part(?, '-', 2) AS bigint)),
        ?, ?, ?, ?
      )
      ON CONFLICT (batch_id, custom_id) DO NOTHING
    columns:
      - customId
      - customId
      - type
      - outputText
      - errorMessage

  - id: apply_results
    type: io.kestra.plugin.jdbc.postgresql.Queries
    description: |
      Succeeded with a valid label: done. Succeeded with any other output: needs_review, raw output kept in enrichment_error.
      Any other result type (errored, expired, canceled or unknown): back to pending with one more attempt, or failed once max_attempts is reached.
      Tickets of this execution still submitted, for example without a result: back to pending.
    url: "{{ secret('POSTGRES_URL') }}"
    username: "{{ secret('POSTGRES_USER') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    sql: |
      UPDATE support_tickets AS t
      SET enrichment_status = 'done',
          category = lower(btrim(r.output_text, E' \t\r\n')),
          enrichment_error = NULL,
          enriched_at = now()
      FROM ticket_enrichment_results AS r
      WHERE r.batch_id = :batch_id
        AND r.custom_id = 'ticket-' || t.id
        AND t.enrichment_execution_id = :execution_id
        AND t.enrichment_status = 'submitted'
        AND r.result_type = 'succeeded'
        AND lower(btrim(r.output_text, E' \t\r\n')) = ANY (string_to_array(:labels, ','));

      UPDATE support_tickets AS t
      SET enrichment_status = 'needs_review',
          enrichment_error = r.output_text
      FROM ticket_enrichment_results AS r
      WHERE r.batch_id = :batch_id
        AND r.custom_id = 'ticket-' || t.id
        AND t.enrichment_execution_id = :execution_id
        AND t.enrichment_status = 'submitted'
        AND r.result_type = 'succeeded';

      UPDATE support_tickets AS t
      SET enrichment_attempts = t.enrichment_attempts + 1,
          enrichment_status = CASE
            WHEN t.enrichment_attempts + 1 >= CAST(:max_attempts AS integer) THEN 'failed'
            ELSE 'pending'
          END,
          enrichment_error = coalesce(r.error_message, r.result_type),
          claimed_at = NULL
      FROM ticket_enrichment_results AS r
      WHERE r.batch_id = :batch_id
        AND r.custom_id = 'ticket-' || t.id
        AND t.enrichment_execution_id = :execution_id
        AND t.enrichment_status = 'submitted'
        AND r.result_type <> 'succeeded';

      UPDATE support_tickets
      SET enrichment_status = 'pending',
          claimed_at = NULL
      WHERE enrichment_execution_id = :execution_id
        AND enrichment_status = 'submitted';
    parameters:
      batch_id: "{{ outputs.create_batch.batchId }}"
      execution_id: "{{ execution.id }}"
      labels: "{{ vars.labels }}"
      max_attempts: "{{ inputs.max_attempts }}"

  - id: summarize
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: Counts this run's tickets by status for the Slack message.
    url: "{{ secret('POSTGRES_URL') }}"
    username: "{{ secret('POSTGRES_USER') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    sql: |
      SELECT count(*) FILTER (WHERE enrichment_status = 'done') AS done,
             count(*) FILTER (WHERE enrichment_status = 'needs_review') AS needs_review,
             count(*) FILTER (WHERE enrichment_status = 'pending') AS pending,
             count(*) FILTER (WHERE enrichment_status = 'failed') AS failed
      FROM support_tickets
      WHERE enrichment_execution_id = :execution_id
    parameters:
      execution_id: "{{ execution.id }}"
    fetchType: FETCH_ONE

  - id: notify
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Posts the run summary to Slack.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    messageText: "Ticket enrichment batch {{ outputs.create_batch.batchId }} ended:
      {{ outputs.summarize.row.done }} done, {{
      outputs.summarize.row.needs_review }} need review, {{
      outputs.summarize.row.pending }} back to pending, {{
      outputs.summarize.row.failed }} failed."

errors:
  - id: cancel_batch
    type: io.kestra.plugin.anthropic.MessageBatch
    description: Stops the batch so unfinished requests are not billed. Requests
      that already finished are billed, and a later run sends their tickets
      again.
    runIf: "{{ outputs.create_batch.batchId is defined }}"
    allowFailure: true
    apiKey: "{{ secret('ANTHROPIC_API_KEY') }}"
    mode: CANCEL
    batchId: "{{ outputs.create_batch.batchId }}"

  - id: release_claims
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: Returns this run's claimed tickets to pending.
    url: "{{ secret('POSTGRES_URL') }}"
    username: "{{ secret('POSTGRES_USER') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    sql: |
      UPDATE support_tickets
      SET enrichment_status = 'pending',
          claimed_at = NULL
      WHERE enrichment_execution_id = :execution_id
        AND enrichment_status = 'submitted'
    parameters:
      execution_id: "{{ execution.id }}"
    fetchType: NONE

  - id: notify_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Posts the failure to Slack.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    messageText: "Ticket enrichment execution {{ execution.id }} failed (batch: {{
      outputs.create_batch.batchId ?? 'none' }}). Its claimed tickets are back
      to pending."

triggers:
  - id: pending_rows
    type: io.kestra.plugin.jdbc.postgresql.Trigger
    description: Disabled by default. Starts a run when at least 100 tickets are
      claimable or the oldest claimable ticket is more than 1 hour old.
    disabled: true
    interval: PT5M
    url: "{{ secret('POSTGRES_URL') }}"
    username: "{{ secret('POSTGRES_USER') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    sql: |
      SELECT count(*) AS claimable, min(created_at) AS oldest_created_at
      FROM support_tickets
      WHERE enrichment_status = 'pending'
         OR (enrichment_status = 'submitted' AND claimed_at < now() - interval '25 hours')
      HAVING count(*) >= 100 OR min(created_at) < now() - interval '1 hour'
    fetchType: FETCH_ONE

Label every pending support ticket in a PostgreSQL table with Claude through the Anthropic Message Batches API, which bills all usage at 50% of the standard API price. Each run claims a slice of tickets, submits them as one batch, waits for the batch to end, writes a category back to each ticket and posts a summary to Slack. Claims, retries, a size budget and an approval gate make it safe to run unattended on a large backlog.

How it works

size_check counts claimable tickets: pending, or submitted by a run that died more than 25 hours ago. With none, nothing_to_claim ends the run.
if_large_batch pauses at approve_batch when more than approval_threshold tickets are claimable and no run was approved in the last 24 hours. After a resume, record_approval stores a 24 hour approval key in the KV store. A pause nobody resumes cancels the run after 24 hours.
claim marks tickets submitted, oldest first, with FOR UPDATE SKIP LOCKED, so overlapping runs never take the same ticket. It stops at max_batch_size tickets or max_batch_bytes of request JSON and returns one request per ticket. If another run took them first, none_claimed ends the run.
create_batch (io.kestra.plugin.anthropic.MessageBatch) submits the requests, and record_batch stores the batch id on each claimed ticket.
wait_for_batch (io.kestra.plugin.core.flow.LoopUntil) runs retrieve_batch every minute until the batch ends, for at most 24 hours 30 minutes; the last check stores the results in an ION file.
load_results (io.kestra.plugin.jdbc.postgresql.Batch) copies every result into ticket_enrichment_results. apply_results updates the tickets in one transaction: a label from labels marks the ticket done, any other answer needs_review, and an error or expiry sends it back to pending, or to failed once max_attempts is reached.
summarize counts this run's tickets and notify posts the counts to Slack.
If a task fails, the errors branch cancels the batch, returns the run's claimed tickets to pending and posts a failure message.
The pending_rows trigger (io.kestra.plugin.jdbc.postgresql.Trigger, disabled by default) checks every 5 minutes and starts a run when at least 100 tickets are claimable or the oldest has waited more than an hour. It never starts a second run while one is going.

What you get

Classification of a whole table at batch prices, without hand-written polling or retry code.
One claim per ticket, even when runs overlap, and recovery from runs that died.
A review state for answers outside your labels, instead of bad values in category.
A human approval before an unusually large backlog is sent.

Who it's for

Support and operations teams labeling tickets, emails or feedback stored in Postgres.
Data engineers who need LLM enrichment of database rows at batch prices.

Why orchestrate this with Kestra

A batch can take up to 24 hours to end. Kestra waits for it without holding a worker, keeps the claim, approval and retry state visible in each execution, cleans up through the errors branch when a step fails, and starts runs from the database with the JDBC trigger.

Prerequisites

An Anthropic API key with access to the Message Batches API.
A Slack incoming webhook.
plugin-anthropic 1.7.0 or later (MessageBatch), plugin-jdbc-postgres and plugin-slack.
PostgreSQL with these two tables:

CREATE TABLE support_tickets (
    id                      bigserial PRIMARY KEY,
    subject                 text NOT NULL,
    body                    text NOT NULL,
    created_at              timestamptz NOT NULL DEFAULT now(),
    category                text,
    enrichment_status       text NOT NULL DEFAULT 'pending'
        CHECK (enrichment_status IN ('pending', 'submitted', 'done', 'needs_review', 'failed')),
    enrichment_attempts     integer NOT NULL DEFAULT 0,
    claimed_at              timestamptz,
    enrichment_batch_id     text,
    enrichment_execution_id text,
    enrichment_error        text,
    enriched_at             timestamptz
);

CREATE TABLE ticket_enrichment_results (
    batch_id      text,
    custom_id     text,
    result_type   text NOT NULL,
    output_text   text,
    error_message text,
    loaded_at     timestamptz NOT NULL DEFAULT now(),
    PRIMARY KEY (batch_id, custom_id)
);

Secrets

ANTHROPIC_API_KEY: creates, polls and cancels the batch.
POSTGRES_URL: JDBC URL of the database, for example jdbc:postgresql://db:5432/app.
POSTGRES_USER and POSTGRES_PASSWORD: a role that can read and update both tables.
SLACK_WEBHOOK_URL: incoming webhook for the summary and failure messages.

Inputs

model (STRING, default claude-haiku-5-5): Claude model for every request.
max_batch_size (INT, default 10000): most tickets one run claims.
max_batch_bytes (INT, default 600000, minimum 50000): most bytes of request JSON one run builds.
approval_threshold (INT, default 1000): claimable tickets above which a run pauses for approval.
max_attempts (INT, default 3): failed results allowed per ticket before it is marked failed.

Quick start

Create the two tables and insert a few tickets.
Add the secrets to the namespace.
Execute the flow, then check category and enrichment_status in support_tickets and the Slack message.
Change labels under variables, and the prompt parameter of claim, to fit your categories.
Enable the pending_rows trigger to label new tickets automatically.

Expected outputs

support_tickets: category and enrichment_status per ticket; enrichment_error holds the answer or error behind needs_review, pending or failed.
ticket_enrichment_results: one row per request, with result_type, output_text and error_message.
{{ outputs.create_batch.batchId }}: the Anthropic batch id.
{{ outputs.retrieve_batch.uri }}: ION file with one row per request (customId, type, outputText, stopReason, errorMessage, rawResponse).
{{ outputs.summarize.row }}: this run's done, needs_review, pending and failed counts, as posted to Slack.

Things to know

Kestra limits each task result message to 1 MiB by default (kestra.queue.message-protection.limit), and claim returns every request in its result. Keep max_batch_bytes well below that limit; a larger backlog simply takes more runs.
If a run fails after its batch was created, the batch is canceled and its tickets are released. Requests that had already finished are still billed, and a later run sends those tickets again.
To retry tickets marked failed, for example after fixing the model name: UPDATE support_tickets SET enrichment_status = 'pending', enrichment_attempts = 0 WHERE enrichment_status = 'failed';

Tested end to end

On Kestra 2.0.5 OSS with Postgres 18 and claude-haiku-5-5.

Run	Result
20 hand-written tickets	20 of 20 labeled correctly
2,020 tickets	4 runs, every ticket `done`, $0.035 in total
Over 1 MiB of requests in one run	Kestra rejects the task result and the tickets are released; `max_batch_bytes` prevents it
Backlog above `approval_threshold`	Pauses; after one approval, runs for 24 hours without pausing
Invalid model	Every request errors and nothing is billed; tickets end `failed` after `max_attempts`
Invalid API key, or wait limit reached	Batch canceled if one exists, tickets released, failure message
Claims 26 hours and 1 hour old	The old ones are taken over, the recent ones left alone
Two runs at once	Disjoint claims, no ticket labeled twice
Trigger	Fires on the age rule and on the count rule, never while its run is going

How to extend

Classify something else, such as priority, language or sentiment, by changing labels and the prompt.
Route needs_review tickets to a Slack channel or a review queue.
Raise max_batch_bytes together with kestra.queue.message-protection.limit to send bigger batches.

Classify Postgres Support Tickets with Claude Message Batches

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Quick start

Expected outputs

Things to know

Tested end to end

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Classify Postgres Support Tickets with Claude Message Batches

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Quick start

Expected outputs

Things to know

Tested end to end

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More AI Blueprints

Accounts Payable 3-Way Match with DuckDB, AI Triage, and Approval Gate

Automated Document RAG Ingestion with Qdrant Vector DB & Slack Notification

Match Newly Exploited CVEs from CISA KEV to Your Stack and Triage Them with AI

Turn an Apify web crawl into a fresh RAG vector index

Build a Chat-With-Your-Data System Using Elasticsearch and OpenAI (RAG)

Chroma Collection Distance Metric Validator

More Data Blueprints

Generate and Execute SQL Queries from Natural Language Using OpenAI GPT-4o

AI-Powered Airflow DAG to Kestra Flow Migration Agent

Airflow DAG Run SLA Gate

Build an ETL Pipeline with Airbyte Cloud and dbt Core for SaaS Analytics

Run Parallel ETL Ingestion with Airbyte to Sync Multiple Data Sources

Run a Scheduled ETL Sync with Airbyte Cloud

New to Kestra?