DEV Community

Mozi Mohidien
Mozi Mohidien

Posted on

Building an AI Enrichment Pipeline with Airbyte and Claude

A while back I was working on a client project where GitHub issues needed to be triaged automatically. The team had hundreds of open tickets — bugs, feature requests, questions, random noise — and wanted a dashboard that grouped them by priority before any human touched them.

My first instinct was the obvious one: write a script. Hit the GitHub API, send each issue to Claude, write results to a table. Done in two hours. Worked great until GitHub rate-limited me on the second run, the schema drifted, they added a second repo, and I was suddenly maintaining three separate scripts that all did mostly the same thing.

So I rebuilt it with Airbyte handling the sync and a small Python enrichment layer doing the AI classification. That combination has held up much better, and this is how to set it up.

Working code: github.com/Mozi1185/airbyte-claude-enrichment

What you'll build:

  • Airbyte connection: GitHub issues → Postgres (incremental sync)
  • Python enricher: reads unprocessed rows, classifies them with claude-haiku-4-5, writes structured results back
  • SQL queries you can actually use

Time: ~45 minutes

Prerequisites: Python 3.10+, Docker (for local Postgres), an Airbyte Cloud account (free tier works), an Anthropic API key, a GitHub personal access token


Why not just write one script

You could call the GitHub API and Claude in the same script. Most tutorials show exactly that. The problem is everything that happens after it works the first time.

GitHub rate-limits you. Claude returns something your parser didn't expect. You add a second repo. A colleague wants to add Jira. Now you're maintaining glue code instead of the actual thing you were trying to build.

Splitting the work cleanly: Airbyte handles connector maintenance, auth refresh, pagination, incremental tracking, schema evolution. That's genuinely annoying to implement correctly and I don't want to do it. The enrichment layer stays stateless and portable — run it on a cron, trigger it from a webhook, drop it into Airflow later. Failed enrichments don't block syncs. Sync failures don't corrupt enriched output. When the client adds Zendesk, I add an Airbyte connection and point the same enricher at a different table.


Step 1: Stand up Postgres

If you already have a Postgres instance, skip this. For local development:

docker run -d \
  --name enrichment-db \
  -e POSTGRES_USER=dev \
  -e POSTGRES_PASSWORD=localdev \
  -e POSTGRES_DB=enrichment \
  -p 5432:5432 \
  postgres:16
Enter fullscreen mode Exit fullscreen mode

Verify it started:

docker exec -it enrichment-db psql -U dev -d enrichment -c "SELECT version();"
Enter fullscreen mode Exit fullscreen mode

Step 2: Connect GitHub to Postgres via Airbyte

Log into Airbyte Cloud. Create a new connection.

Source: GitHub

  1. Go to Sources → + New source → GitHub
  2. Enter the repository — e.g. airbytehq/airbyte or your own
  3. Paste your GitHub personal access token (repo scope — generate at github.com/settings/tokens)
  4. Click Set up source

Destination: Postgres

  1. Go to Destinations → + New destination → Postgres
  2. Fill in your connection details:
    • Host: localhost
    • Port: 5432
    • Database: enrichment
    • Username: dev
    • Password: localdev
    • Default Schema: github_raw
  3. Click Set up destination

Heads up: If you're running Postgres locally, Airbyte Cloud can't reach localhost — they're on different networks. I burned an afternoon on this before realizing it. The fix is ngrok: ngrok tcp 5432. Copy the tunnel host and port (something like 0.tcp.ngrok.io:12345) and use those in the destination config instead of localhost.

Connection settings

  1. Under Streams, select issues only — uncheck everything else
  2. Sync mode: Incremental | Append + Deduped
  3. Schedule: every 6 hours (or trigger manually for now)
  4. Click Set up connection, then Sync now

The first sync creates github_raw.issues in Postgres and populates it. Verify:

SELECT id, title, state, created_at
FROM github_raw.issues
LIMIT 5;
Enter fullscreen mode Exit fullscreen mode

The body column is what Claude will read. If it's empty on some rows, that's normal — GitHub allows issues without a body.


Step 3: Create the enrichment table

CREATE TABLE IF NOT EXISTS github_raw.issues_enriched (
    issue_id         BIGINT PRIMARY KEY,
    category         TEXT,
    priority         TEXT,
    one_line_summary TEXT,
    enriched_at      TIMESTAMPTZ DEFAULT now()
);
Enter fullscreen mode Exit fullscreen mode

PRIMARY KEY on issue_id makes re-runs safe — it upserts rather than duplicating rows.


Step 4: Install dependencies

pip install anthropic psycopg2-binary python-dotenv
Enter fullscreen mode Exit fullscreen mode

Create .env:

ANTHROPIC_API_KEY=sk-ant-...
DB_HOST=localhost
DB_PORT=5432
DB_NAME=enrichment
DB_USER=dev
DB_PASSWORD=localdev
Enter fullscreen mode Exit fullscreen mode

Step 5: Write the enricher

# enricher.py
import os
import json
import psycopg2
import anthropic
from dotenv import load_dotenv

load_dotenv()

client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

DB_CONFIG = {
    "host":     os.environ["DB_HOST"],
    "port":     int(os.environ["DB_PORT"]),
    "dbname":   os.environ["DB_NAME"],
    "user":     os.environ["DB_USER"],
    "password": os.environ["DB_PASSWORD"],
}

SYSTEM_PROMPT = """You are a triage assistant for a software project.
Given a GitHub issue title and body, return a JSON object with exactly these fields:
- category: one of "bug", "feature_request", "documentation", "question", "other"
- priority: one of "critical", "high", "medium", "low"
- one_line_summary: a single sentence (max 120 chars) describing the issue

Return only valid JSON. No explanation, no markdown fences."""


def classify_issue(title: str, body: str) -> dict:
    """Send one issue to Claude and return structured classification."""
    content = f"Title: {title}\n\nBody:\n{body or '(no body provided)'}"

    message = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=256,
        system=SYSTEM_PROMPT,
        messages=[{"role": "user", "content": content}],
    )

    raw = message.content[0].text.strip()
    return json.loads(raw)


def get_unenriched_issues(conn, limit: int = 50) -> list[dict]:
    """Fetch issues that haven't been enriched yet."""
    with conn.cursor() as cur:
        cur.execute("""
            SELECT i.id, i.title, i.body
            FROM github_raw.issues i
            LEFT JOIN github_raw.issues_enriched e ON e.issue_id = i.id
            WHERE e.issue_id IS NULL
              AND i.state = 'open'
            ORDER BY i.created_at DESC
            LIMIT %s
        """, (limit,))
        cols = [desc[0] for desc in cur.description]
        return [dict(zip(cols, row)) for row in cur.fetchall()]


def write_enrichment(conn, issue_id: int, result: dict) -> None:
    """Upsert enrichment result for one issue."""
    with conn.cursor() as cur:
        cur.execute("""
            INSERT INTO github_raw.issues_enriched
                (issue_id, category, priority, one_line_summary, enriched_at)
            VALUES (%s, %s, %s, %s, now())
            ON CONFLICT (issue_id) DO UPDATE SET
                category         = EXCLUDED.category,
                priority         = EXCLUDED.priority,
                one_line_summary = EXCLUDED.one_line_summary,
                enriched_at      = now()
        """, (
            issue_id,
            result["category"],
            result["priority"],
            result["one_line_summary"],
        ))
    conn.commit()


def run_enrichment_batch(limit: int = 50) -> None:
    conn = psycopg2.connect(**DB_CONFIG)
    try:
        issues = get_unenriched_issues(conn, limit)
        print(f"Found {len(issues)} unenriched issues")

        for issue in issues:
            issue_id = issue["id"]
            try:
                result = classify_issue(
                    issue["title"] or "",
                    issue["body"] or "",
                )
                write_enrichment(conn, issue_id, result)
                print(f"  #{issue_id}{result['category']} / {result['priority']}")
            except json.JSONDecodeError as e:
                print(f"  #{issue_id} parse error: {e} — skipping")
            except anthropic.APIError as e:
                print(f"  #{issue_id} API error: {e} — skipping")
    finally:
        conn.close()


if __name__ == "__main__":
    run_enrichment_batch()
Enter fullscreen mode Exit fullscreen mode

Run it:

python enricher.py
Enter fullscreen mode Exit fullscreen mode

Expected output:

Found 47 unenriched issues
  #1823 → bug / high
  #1801 → feature_request / medium
  #1799 → documentation / low
  #1788 → question / low
  ...
Enter fullscreen mode Exit fullscreen mode

Step 6: Query the results

High-priority bugs, newest first:

SELECT
    i.id,
    i.title,
    e.category,
    e.priority,
    e.one_line_summary,
    i.created_at
FROM github_raw.issues i
JOIN github_raw.issues_enriched e ON e.issue_id = i.id
WHERE e.category = 'bug'
  AND e.priority IN ('critical', 'high')
ORDER BY i.created_at DESC
LIMIT 20;
Enter fullscreen mode Exit fullscreen mode

Full breakdown across categories:

SELECT
    e.category,
    e.priority,
    COUNT(*) AS issue_count
FROM github_raw.issues_enriched e
GROUP BY e.category, e.priority
ORDER BY e.category, e.priority;
Enter fullscreen mode Exit fullscreen mode

Step 7: Keep it running

The enricher is stateless — run it on any scheduler. Simplest option:

# Cron: run enrichment 15 minutes after each Airbyte sync (syncs at :00)
15 */6 * * * cd /path/to/project && python enricher.py >> enricher.log 2>&1
Enter fullscreen mode Exit fullscreen mode

If you want tighter coupling, Airbyte has Webhook Notifications under each connection's settings. Point the webhook at an endpoint that calls run_enrichment_batch() and the enricher fires exactly when a sync completes — no delay, no polling.


Eliminating parse errors with tool use

The enricher above parses Claude's JSON from raw text. It works most of the time. But Claude occasionally wraps the output in a markdown code fence, or adds a clarifying sentence before the JSON, and then json.loads blows up. The json.JSONDecodeError catch in the loop handles this by skipping the row — which is fine, but skipped rows means gaps in your enrichment.

The cleaner fix: Claude's tool use API. Instead of asking for JSON in the prompt, you define a tool with the exact schema and force Claude to call it. The response comes back already structured — no parsing step.

CLASSIFY_TOOL = {
    "name": "classify_issue",
    "description": "Classify a GitHub issue by category, priority, and summary.",
    "input_schema": {
        "type": "object",
        "properties": {
            "category": {
                "type": "string",
                "enum": ["bug", "feature_request", "documentation", "question", "other"]
            },
            "priority": {
                "type": "string",
                "enum": ["critical", "high", "medium", "low"]
            },
            "one_line_summary": {
                "type": "string",
                "description": "Single sentence, max 120 characters"
            }
        },
        "required": ["category", "priority", "one_line_summary"]
    }
}


def classify_issue_structured(title: str, body: str) -> dict:
    """Classify using tool use — no JSON parsing, no parse errors."""
    content = f"Title: {title}\n\nBody:\n{body or '(no body provided)'}"

    message = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=256,
        tools=[CLASSIFY_TOOL],
        tool_choice={"type": "tool", "name": "classify_issue"},
        messages=[{"role": "user", "content": content}],
    )

    # content[0] is always a ToolUseBlock when tool_choice forces a specific tool
    return message.content[0].input  # already a dict
Enter fullscreen mode Exit fullscreen mode

Swap classify_issue for classify_issue_structured in run_enrichment_batch and remove the json.JSONDecodeError catch — you won't hit it anymore.


What this costs

claude-haiku-4-5 is $0.80 per million input tokens and $4 per million output tokens. Each issue uses roughly 300–600 input tokens and about 80 output tokens.

At 10,000 issues:

Component Tokens Cost
Input (avg 450 × 10k) 4.5M ~$3.60
Output (avg 80 × 10k) 0.8M ~$3.20
Total ~$6.80

That's for the initial backfill. After that, Airbyte only syncs new rows since the last run, so ongoing enrichment costs scale with new volume only — not the full history.

If you want to cut input costs further, use prompt caching. The system prompt is identical across every call, so caching it drops input costs by roughly 90% on repeated runs:

message = client.messages.create(
    model="claude-haiku-4-5",
    max_tokens=256,
    system=[{
        "type": "text",
        "text": SYSTEM_PROMPT,
        "cache_control": {"type": "ephemeral"}
    }],
    messages=[{"role": "user", "content": content}],
)
Enter fullscreen mode Exit fullscreen mode

Worth doing if you're running this on a large repo.


Swapping sources

The enrichment code doesn't know anything about GitHub. It reads from a table Airbyte created. To switch to Jira, add a new Airbyte connection with Jira as the source and the same Postgres destination. Airbyte creates a new schema and table. Write a get_unenriched_jira_issues function pointing at that table. The classify_issue function is reusable as-is.

Airbyte has 300+ connectors — Zendesk, HubSpot, Salesforce, Notion, Linear. The enrichment pattern applies to all of them. That's the point of keeping the sync and enrichment separate.


What you end up with: a Postgres table populated by Airbyte on a schedule, an enrichment layer that classifies each new row after each sync, and structured output queryable by whatever SQL tool your team uses. The sync and enrichment stay independent — neither can break the other, and adding a new source is an Airbyte config change, not a code change.


Mozi Mohidien builds AI automation systems — Claude API, Airbyte, n8n, Telegram. Writing at dev.to/mozimohidien.

Top comments (0)