bootstrapping-agent

Bootstrapping an Agent with an Airbyte Connector

Install the SDK

uv pip install airbyte-agent-sdk

The single airbyte-agent-sdk package ships every typed connector. Import them from airbyte_agent_sdk.connectors.{slug}. tool_utils, list_entities(), and entity_schema() are only available on typed connectors.

Core Pattern (PydanticAI)

import os
from pydantic_ai import Agent
from airbyte_agent_sdk import AirbyteAuthConfig
from airbyte_agent_sdk.connectors.stripe import StripeConnector

connector = StripeConnector(
    auth_config=AirbyteAuthConfig(
        airbyte_client_id=os.getenv("AIRBYTE_CLIENT_ID"),
        airbyte_client_secret=os.getenv("AIRBYTE_CLIENT_SECRET"),
        workspace_name=os.getenv("AIRBYTE_WORKSPACE_NAME", "default"),
    )
)

agent = Agent(
    "<provider:model>",
    system_prompt=(
        "You are a helpful assistant with access to Stripe. "
        "Use the stripe_execute tool to look up customer, invoice, and balance data. "
        "Ask for clarification if a request is ambiguous."
    ),
)

@agent.tool_plain
@StripeConnector.tool_utils
async def stripe_execute(entity: str, action: str, params: dict | None = None):
    return await connector.execute(entity, action, params or {})

Always hosted mode: Use AirbyteAuthConfig with airbyte_client_id and airbyte_client_secret. Never generate local auth code.

Decorator Stacking

The framework decorator goes on top, tool_utils goes underneath:

@agent.tool_plain           # Framework registers this as a tool
@StripeConnector.tool_utils # Enriches docstring with connector capabilities
async def stripe_execute(...):

tool_utils is a @classmethod — use StripeConnector.tool_utils, not connector.tool_utils.

Automatic Retry Translation

tool_utils automatically translates retryable errors to the framework's retry signal (ModelRetry for pydantic-ai). The example above continues to work unchanged — translation happens inside tool_utils with no extra decorator needed.