Airbyte Agent Connectors

Terminology: Platform Mode = Airbyte Cloud at app.airbyte.ai (managed credentials, UI visibility). OSS Mode = local Python SDK (self-managed credentials, no cloud dependency). Definition ID = UUID that identifies a connector type in the Airbyte API (used in

definition_id

fields, not

connector_type

or

connector_definition_id

).

Important: This skill provides documentation and setup guidance. When helping users set up connectors, follow the documented workflows below. Do NOT attempt to import Python modules, verify package installations, or run code to check configurations -- simply guide users through the steps using the code examples provided.

Mode Detection

Platform Mode (Airbyte Cloud)

• Environment has

  AIRBYTE_CLIENT_ID

  +

  AIRBYTE_CLIENT_SECRET

• User wants connectors visible in the Airbyte UI at app.airbyte.ai
• User needs managed credential storage, context store, or multi-tenant deployments

See Agent Engine Platform docs and the hosted quickstart tutorial for full details.

OSS Mode (Open Source / Local SDK)

• User wants to run connectors directly without platform integration
• User is doing quick development or prototyping
• User wants Claude Code/Desktop integration via MCP only

Ask if unclear: "Are you using Airbyte Platform (app.airbyte.ai) or open source connectors?"

Supported Connectors

connector.execute(entity, action, params)

airbyte-agent-stripe

airbyte-agent-hubspot

airbyte-agent-github

airbyte-agent-salesforce

airbyte-agent-gong

Full table: See references/connector-index.md for all 51 connectors with auth types, key entities, and documentation status.

Platform Mode Quick Start

Prerequisites

• AIRBYTE_CLIENT_ID

• AIRBYTE_CLIENT_SECRET

Create a Connector

from airbyte_agent_stripe import StripeConnector
from airbyte_agent_stripe.models import StripeAuthConfig

connector = await StripeConnector.create_hosted(
    external_user_id="user_123",
    airbyte_client_id="...",
    airbyte_client_secret="...",
    auth_config=StripeAuthConfig(api_key="sk_live_...")
)

Use Existing Connector

connector = StripeConnector(
    external_user_id="user_123",
    airbyte_client_id="...",
    airbyte_client_secret="...",
)
result = await connector.execute("customers", "list", {"limit": 10})

OSS Mode Quick Start

Install

# Using uv (recommended)
uv add airbyte-agent-github

# Or using pip in a virtual environment
python3 -m venv .venv && source .venv/bin/activate
pip install airbyte-agent-github

Use Directly

from airbyte_agent_github import GithubConnector
from airbyte_agent_github.models import GithubPersonalAccessTokenAuthConfig

connector = GithubConnector(
    auth_config=GithubPersonalAccessTokenAuthConfig(token="ghp_your_token")
)

result = await connector.execute("issues", "list", {
    "owner": "airbytehq",
    "repo": "airbyte",
    "states": ["OPEN"],
    "per_page": 10
})

Add to Claude via MCP

claude mcp add airbyte-agent-mcp --scope project

Entity-Action API Pattern

result = await connector.execute(entity, action, params)
# For list actions: returns an envelope with result.data (list) and result.meta (pagination dict)
# For get/create/update actions: returns a raw dict (use dict access like result['name'])
# Raises exceptions on failure (RuntimeError, TypeError, ValueError, etc.)

Important: Error handling in agent tools. When registering

execute()

as a tool for an AI agent (PydanticAI, LangChain, etc.), always wrap the call in try/except and return the error message as a string. This prevents unhandled exceptions from crashing the agent and lets the LLM recover gracefully:

python

@agent.tool_plain
async def execute(entity: str, action: str, params: dict | None = None) -> str:
    try:
        return await connector.execute(entity, action, params or {})
    except Exception as e:
        return f"Error: {type(e).__name__}: {e}"

tool_utils

and

enable_hosted_mode_features

: The

@Connector.tool_utils

decorator auto-generates the tool's docstring from the connector schema, including available entities, actions, parameters, and usage guidelines. In Platform Mode (default), the docstring includes

search

actions that query the context store and guidance to prefer cached search over direct API calls. In OSS Mode, pass

enable_hosted_mode_features=False

to exclude these — the context store is not available locally, so advertising search actions would cause errors:

python

# OSS Mode — excludes context store search actions from the tool docstring
@Connector.tool_utils(enable_hosted_mode_features=False)

# Platform Mode (default) — includes search actions and context store guidance
@Connector.tool_utils

Actions

result.data

list

list[dict]

get

dict

create

dict

update

dict

delete

dict

api_search

list[dict]

Quick Examples

# List
await connector.execute("customers", "list", {"limit": 10})

# Get
await connector.execute("customers", "get", {"id": "cus_xxx"})

# Search
await connector.execute("repositories", "api_search", {
    "query": "language:python stars:>1000"
})

Pagination

async def fetch_all(connector, entity, params=None):
    all_records = []
    cursor = None
    params = params or {}

    while True:
        if cursor:
            params["after"] = cursor
        result = await connector.execute(entity, "list", params)
        all_records.extend(result.data)

        if result.meta and hasattr(result.meta, 'pagination'):
            cursor = getattr(result.meta.pagination, 'cursor', None)
            if not cursor:
                break
        else:
            break

    return all_records

Authentication Quick Reference

API Key Connectors

# Stripe
from airbyte_agent_stripe.models import StripeAuthConfig
auth_config=StripeAuthConfig(api_key="sk_live_...")

# Gong
from airbyte_agent_gong.models import GongAccessKeyAuthenticationAuthConfig
auth_config=GongAccessKeyAuthenticationAuthConfig(
    access_key="...", access_key_secret="..."
)

# HubSpot (Private App)
from airbyte_agent_hubspot.models import HubspotPrivateAppAuthConfig
auth_config=HubspotPrivateAppAuthConfig(private_app_token="pat-na1-...")

# HubSpot (OAuth)
from airbyte_agent_hubspot.models import HubspotOauth2AuthConfig
auth_config=HubspotOauth2AuthConfig(
    client_id="...", client_secret="...", refresh_token="..."
)

Personal Access Token

# GitHub
from airbyte_agent_github.models import GithubPersonalAccessTokenAuthConfig
auth_config=GithubPersonalAccessTokenAuthConfig(token="ghp_...")

# Slack
from airbyte_agent_slack.models import SlackAuthConfig
auth_config=SlackAuthConfig(token="xoxb-...")

OAuth (requires refresh token)

# Salesforce
from airbyte_agent_salesforce.models import SalesforceAuthConfig
auth_config=SalesforceAuthConfig(
    client_id="...", client_secret="...", refresh_token="..."
)

Per-connector auth details: Each connector reference in references/connectors/ includes the specific auth class name and fields.

Security Best Practices

• Never hard-code credentials in source files. Use environment variables or

  .env

  files.
• Use

  .env

  pattern:

  from dotenv import load_dotenv; load_dotenv()

• Rotate tokens regularly. Use short-lived tokens where possible.
• For production, use Platform Mode for managed credential storage.

Per-Connector Reference Documentation

• Connector Index -- full table of all connectors
• Per-connector references: references/connectors/

Reference Documentation

External Documentation

How References Are Generated

scripts/generate_skill_references.py

python scripts/generate_skill_references.py --all

--connector <name>

Skill Metadata

• Author: Airbyte
• Version: 1.2.0
• License: Elastic-2.0
• Compatibility: Requires Python 3.11+. Recommends uv for package management.
• Repository: https://github.com/airbytehq/airbyte-agent-connectors
• MCP Server: airbyte-agent-mcp

Support

• Slack Community: slack.airbyte.com
• GitHub Issues: airbytehq/airbyte-agent-connectors
• Documentation: docs.airbyte.com/ai-agents