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_...")
)

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

undefined

undefined

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
})

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

result = await connector.execute(entity, action, params)

> **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](https://docs.airbyte.com/ai-agents/platform/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
> ```

undefined

undefined

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

undefined

undefined

undefined

undefined

undefined

> **Per-connector auth details:** Each connector reference in [references/connectors/](references/connectors/) includes the specific auth class name and fields.

---