Airbyte Management Skill

Manage and monitor Airbyte connections, sync jobs, and data pipeline health via the Airbyte API.

MANDATORY: Discovery-First Pattern

Always list workspaces and connections before querying specific jobs or sources.

Phase 1: Discovery

#!/bin/bash

airbyte_api() {
    local method="${1:-POST}"
    local endpoint="$2"
    local data="${3:-{}}"

    curl -s -X "$method" \
        -H "Content-Type: application/json" \
        -H "Authorization: Bearer ${AIRBYTE_API_TOKEN}" \
        "${AIRBYTE_API_URL}/api/v1/${endpoint}" \
        -d "$data"
}

echo "=== Workspaces ==="
airbyte_api POST "workspaces/list" | jq -r '
    .workspaces[] | "\(.workspaceId)\t\(.name)\t\(.slug)"
' | column -t

echo ""
echo "=== Connections Summary ==="
airbyte_api POST "connections/list" "{\"workspaceId\": \"${AIRBYTE_WORKSPACE_ID}\"}" | jq -r '
    .connections[] | "\(.connectionId[0:8])\t\(.name)\t\(.status)\t\(.scheduleType // "manual")"
' | column -t | head -30

echo ""
echo "=== Source/Destination Count ==="
SOURCES=$(airbyte_api POST "sources/list" "{\"workspaceId\": \"${AIRBYTE_WORKSPACE_ID}\"}" | jq '.sources | length')
DESTS=$(airbyte_api POST "destinations/list" "{\"workspaceId\": \"${AIRBYTE_WORKSPACE_ID}\"}" | jq '.destinations | length')
echo "Sources: $SOURCES | Destinations: $DESTS"

Core Helper Functions

#!/bin/bash

airbyte_api() {
    local method="${1:-POST}"
    local endpoint="$2"
    local data="${3:-{}}"

    curl -s -X "$method" \
        -H "Content-Type: application/json" \
        -H "Authorization: Bearer ${AIRBYTE_API_TOKEN}" \
        "${AIRBYTE_API_URL}/api/v1/${endpoint}" \
        -d "$data"
}

Output Rules

• TOKEN EFFICIENCY: Target ≤50 lines per output
• Airbyte API is mostly POST-based even for reads — always use POST with JSON body
• Never dump full catalog — extract stream names and sync modes only

Common Operations

Connection Health Dashboard

#!/bin/bash
echo "=== Connection Status Overview ==="
airbyte_api POST "connections/list" "{\"workspaceId\": \"${AIRBYTE_WORKSPACE_ID}\"}" | jq '
    .connections | {
        total: length,
        active: [.[] | select(.status == "active")] | length,
        inactive: [.[] | select(.status == "inactive")] | length,
        deprecated: [.[] | select(.status == "deprecated")] | length
    }'

echo ""
echo "=== Recent Job Failures ==="
airbyte_api POST "connections/list" "{\"workspaceId\": \"${AIRBYTE_WORKSPACE_ID}\"}" | jq -r '
    .connections[] | .connectionId
' | head -20 | while read cid; do
    airbyte_api POST "jobs/list" "{\"configTypes\": [\"sync\"], \"configId\": \"${cid}\", \"pagination\": {\"pageSize\": 1}}" | jq -r '
        .jobs[]? | select(.job.status == "failed") |
        "\(.job.configId[0:8])\t\(.job.status)\t\(.job.createdAt | todate)\t\(.attempts[-1].failureSummary?.failures[0]?.externalMessage[0:60] // "unknown")"
    '
done | column -t | head -15

Sync Job Details

#!/bin/bash
CONNECTION_ID="${1:?Connection ID required}"

echo "=== Connection Details ==="
airbyte_api POST "connections/get" "{\"connectionId\": \"${CONNECTION_ID}\"}" | jq '{
    name: .name,
    status: .status,
    schedule: .scheduleType,
    namespace: .namespaceFormat,
    prefix: .prefix,
    streams: (.syncCatalog.streams | length)
}'

echo ""
echo "=== Recent Sync Jobs ==="
airbyte_api POST "jobs/list" "{\"configTypes\": [\"sync\"], \"configId\": \"${CONNECTION_ID}\", \"pagination\": {\"pageSize\": 10}}" | jq -r '
    .jobs[] | "\(.job.id)\t\(.job.status)\t\(.job.createdAt | todate)\t\(.job.updatedAt | todate)"
' | column -t

echo ""
echo "=== Last Sync Stats ==="
airbyte_api POST "jobs/list" "{\"configTypes\": [\"sync\"], \"configId\": \"${CONNECTION_ID}\", \"pagination\": {\"pageSize\": 1}}" | jq '
    .jobs[0].attempts[-1].attempt | {
        status: .status,
        records_synced: .totalStats.recordsEmitted,
        bytes_synced: .totalStats.bytesEmitted,
        records_committed: .totalStats.recordsCommitted
    }'

Source and Destination Health

#!/bin/bash
echo "=== Sources ==="
airbyte_api POST "sources/list" "{\"workspaceId\": \"${AIRBYTE_WORKSPACE_ID}\"}" | jq -r '
    .sources[] | "\(.sourceId[0:8])\t\(.sourceName)\t\(.name)\t\(.connectionConfiguration | keys | join(","))"
' | column -t | head -20

echo ""
echo "=== Destinations ==="
airbyte_api POST "destinations/list" "{\"workspaceId\": \"${AIRBYTE_WORKSPACE_ID}\"}" | jq -r '
    .destinations[] | "\(.destinationId[0:8])\t\(.destinationName)\t\(.name)"
' | column -t | head -10

echo ""
echo "=== Source Health Check ==="
SOURCE_ID="${1:-}"
if [ -n "$SOURCE_ID" ]; then
    airbyte_api POST "sources/check_connection" "{\"sourceId\": \"${SOURCE_ID}\"}" | jq '{
        status: .status,
        message: .message
    }'
fi

Schema Change Detection

#!/bin/bash
CONNECTION_ID="${1:?Connection ID required}"

echo "=== Configured Streams ==="
airbyte_api POST "connections/get" "{\"connectionId\": \"${CONNECTION_ID}\"}" | jq -r '
    .syncCatalog.streams[] |
    "\(.stream.name)\t\(.config.syncMode)\t\(.config.destinationSyncMode)\t\(.stream.jsonSchema.properties | keys | length) cols"
' | column -t | head -20

echo ""
echo "=== Discover Source Schema (fresh) ==="
SOURCE_ID=$(airbyte_api POST "connections/get" "{\"connectionId\": \"${CONNECTION_ID}\"}" | jq -r '.sourceId')
airbyte_api POST "sources/discover_schema" "{\"sourceId\": \"${SOURCE_ID}\"}" | jq -r '
    .catalog.streams[] | "\(.stream.name)\t\(.stream.jsonSchema.properties | keys | length) cols\t\(.stream.supportedSyncModes | join(","))"
' | column -t | head -20

Trigger and Manage Syncs

#!/bin/bash
CONNECTION_ID="${1:?Connection ID required}"
ACTION="${2:-status}"

case "$ACTION" in
    trigger)
        echo "=== Triggering Sync ==="
        airbyte_api POST "connections/sync" "{\"connectionId\": \"${CONNECTION_ID}\"}" | jq '{
            job_id: .job.id,
            status: .job.status,
            config_type: .job.configType
        }'
        ;;
    reset)
        echo "=== Triggering Reset ==="
        airbyte_api POST "connections/reset" "{\"connectionId\": \"${CONNECTION_ID}\"}" | jq '{
            job_id: .job.id,
            status: .job.status
        }'
        ;;
    status)
        echo "=== Current Status ==="
        airbyte_api POST "connections/get" "{\"connectionId\": \"${CONNECTION_ID}\"}" | jq '{
            name: .name, status: .status, schedule: .scheduleType
        }'
        ;;
esac

Output Format

Present results as a structured report:

Managing Airbyte Report
═══════════════════════
Resources discovered: [count]

Resource       Status    Key Metric    Issues
──────────────────────────────────────────────
[name]         [ok/warn] [value]       [findings]

Summary: [total] resources | [ok] healthy | [warn] warnings | [crit] critical
Action Items: [list of prioritized findings]

Target ≤50 lines of output. Use tables for multi-resource comparisons.

Anti-Hallucination Rules

1. NEVER assume resource names — always discover via CLI/API in Phase 1 before referencing in Phase 2.
2. NEVER fabricate metric names or dimensions — verify against the service documentation or --help output.
3. NEVER mix CLI commands between service versions — confirm which version/API you are targeting.
4. ALWAYS use the discovery → verify → analyze chain — every resource referenced must have been discovered first.
5. ALWAYS handle empty results gracefully — an empty response is valid data, not an error to retry.

Counter-Rationalizations

Shortcut	Counter	Why
"I'll skip discovery and check known resources"	Always run Phase 1 discovery first	Resource names change, new resources appear — assumed names cause errors
"The user only asked for a quick check"	Follow the full discovery → analysis flow	Quick checks miss critical issues; structured analysis catches silent failures
"Default configuration is probably fine"	Audit configuration explicitly	Defaults often leave logging, security, and optimization features disabled
"Metrics aren't needed for this"	Always check relevant metrics when available	API/CLI responses show current state; metrics reveal trends and intermittent issues
"I don't have access to that"	Try the command and report the actual error	Assumed permission failures prevent useful investigation; actual errors are informative

Common Pitfalls

• POST for reads: Airbyte API uses POST for list/get operations — do not use GET for these endpoints
• Workspace scope: All queries require workspaceId — always discover workspaces first
• Connection vs Connector: A "connection" links a source to a destination; a "connector" is the source/destination type definition
• Schema discovery: discover_schema can be slow (minutes) for large databases — it queries the actual source
• Sync modes: full_refresh re-syncs everything; incremental only new/changed — wrong mode causes data loss or duplication
• Destination sync modes: overwrite replaces data; append adds rows; append_dedup upserts — verify before changing
• Reset impact: connections/reset deletes all data in destination and re-syncs — use with extreme caution
• Job attempts: A single job may have multiple attempts on failure — check attempts[-1] for latest
• Normalization: If enabled, Airbyte runs normalization after sync — failures may occur in normalization, not sync