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

Core Pattern (PydanticAI)

Use build_connector_tools — the preferred default on supported frameworks — unless the agent needs custom tool bodies:

import os
from pydantic_ai import Agent
from airbyte_agent_sdk import AirbyteAuthConfig, build_connector_tools
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"), ) )

tools = build_connector_tools(connector, framework="pydantic_ai")

agent = Agent( "provider:model", tools=tools.as_list(), system_prompt=( "You are a helpful assistant with access to Stripe. " "Inspect the connector, read the relevant docs, then use execute to look up data. " "Ask for clarification if a request is ambiguous." ), )


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

## Custom Tool Bodies (`agent_tool`)

When a tool needs a custom body, use `agent_tool` even when the framework is natively supported. Decorate and register execute, inspect, and docs functions so the model can discover connector capabilities progressively:

```python
manual_agent = Agent("<provider:model>")

@manual_agent.tool_plain
@StripeConnector.agent_tool(
    framework="pydantic_ai",
    inspect_tool="stripe_inspect",
    docs_tool="stripe_read_docs",
)
async def stripe_execute(entity: str, action: str, params: dict | None = None):
    return await connector.execute(entity, action, params or {})

@manual_agent.tool_plain
@StripeConnector.agent_tool(framework="pydantic_ai")
async def stripe_inspect():
    return await connector.inspect_connector()

@manual_agent.tool_plain
@StripeConnector.agent_tool(framework="pydantic_ai")
async def stripe_read_docs(section: str | None = None):
    return await connector.read_skill_docs(section)

The framework registration decorator goes on top and agent_tool goes underneath. agent_tool is a @classmethod: use StripeConnector.agent_tool(...), not connector.agent_tool(...). The role is inferred from each signature — (entity, action, ...) → execute, (section, ...) → docs, () → inspect; extra params are allowed, and ambiguous signatures must pass the role explicitly, e.g. agent_tool("execute").

Automatic Retry Translation

Both build_connector_tools(..., framework="pydantic_ai") and agent_tool(framework="pydantic_ai") translate retryable errors to PydanticAI's ModelRetry. Use framework="langchain", "openai_agents", or "mcp" for the corresponding supported framework. No extra translation decorator is needed.

Reference demo: connector-sdk/examples/demo_agent.py (mocked, no credentials needed: --mock).

Unsupported Frameworks

On a framework the SDK does not natively support, use the same three-function agent_tool pattern and omit framework= (role inference works the same as above):

@StripeConnector.agent_tool(inspect_tool="stripe_inspect", docs_tool="stripe_read_docs")
async def stripe_execute(entity: str, action: str, params: dict | None = None):
    return await connector.execute(entity, action, params or {})

@StripeConnector.agent_tool()
async def stripe_inspect():
    return await connector.inspect_connector()

@StripeConnector.agent_tool()
async def stripe_read_docs(section: str | None = None):
    return await connector.read_skill_docs(section)

Register all three with the target framework. The execute docstring steers the model through inspect → docs outline → docs section → execute instead of embedding the full entity/action reference. Failures raise AirbyteToolError (from airbyte_agent_sdk import AirbyteToolError) by default — no framework auto-detection; pass framework="..." to target a supported framework's retry signal instead. The optional inspect_tool=/docs_tool= kwargs put the exact registered sibling-tool names into the execute docstring; omit them for generic phrasing.

Legacy Integrations (`tool_utils`)

Connector.tool_utils is deprecated; it remains available only for existing integrations that use a single broad tool description. Never generate it for new tools. When editing a legacy integration, keep the framework decorator on top and tool_utils underneath unless the task includes migrating it to the three-tool agent_tool flow.

Verify the Setup

check = await connector.check()
if check.status == "healthy":
    print(f"Connected — checked {check.checked_entity}/{check.checked_action}")
else:
    print(f"Failed: {check.error}")

Framework Detection

Detect the developer's framework from their existing imports:

from pydantic_ai import Agent → Use PydanticAI patterns
from anthropic import Anthropic → Use Claude SDK patterns
If unclear, ask which framework they're using

Connector Naming Convention

All connectors ship in airbyte-agent-sdk. Import each one from airbyte_agent_sdk.connectors.{slug}:

Connector	Import	Class
stripe	`airbyte_agent_sdk.connectors.stripe`	`StripeConnector`
zendesk-support	`airbyte_agent_sdk.connectors.zendesk_support`	`ZendeskSupportConnector`
hubspot	`airbyte_agent_sdk.connectors.hubspot`	`HubspotConnector`

Hyphens in connector slugs become underscores in the submodule path.

Environment Variables

The developer needs these in their .env:

AIRBYTE_CLIENT_ID=your_client_id
AIRBYTE_CLIENT_SECRET=your_client_secret
AIRBYTE_WORKSPACE_NAME=your_workspace_name

References

SDK API reference — full API signatures and options
PydanticAI patterns — complete runnable examples
Claude SDK patterns — Anthropic Python SDK examples