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
Verify it started:
docker exec -it enrichment-db psql -U dev -d enrichment -c "SELECT version();"
Step 2: Connect GitHub to Postgres via Airbyte
Log into Airbyte Cloud. Create a new connection.
Source: GitHub
- Go to Sources → + New source → GitHub
- Enter the repository — e.g.
airbytehq/airbyteor your own - Paste your GitHub personal access token (
reposcope — generate at github.com/settings/tokens) - Click Set up source
Destination: Postgres
- Go to Destinations → + New destination → Postgres
- Fill in your connection details:
- Host:
localhost - Port:
5432 - Database:
enrichment - Username:
dev - Password:
localdev - Default Schema:
github_raw
- Host:
- 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 like0.tcp.ngrok.io:12345) and use those in the destination config instead of localhost.
Connection settings
- Under Streams, select
issuesonly — uncheck everything else - Sync mode: Incremental | Append + Deduped
- Schedule: every 6 hours (or trigger manually for now)
- 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;
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()
);
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
Create .env:
ANTHROPIC_API_KEY=sk-ant-...
DB_HOST=localhost
DB_PORT=5432
DB_NAME=enrichment
DB_USER=dev
DB_PASSWORD=localdev
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()
Run it:
python enricher.py
Expected output:
Found 47 unenriched issues
#1823 → bug / high
#1801 → feature_request / medium
#1799 → documentation / low
#1788 → question / low
...
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;
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;
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
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
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}],
)
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)