Wires up an Airbyte connector for use in a PydanticAI or Claude SDK agent, or any other framework via agent_tool. Generates auth config, connector initialization, and tool_utils- or agent_tool-decorated tool functions. Use when adding a connector to an agent or setting up a new agent with a connector.
Wires up an Airbyte connector for use in a PydanticAI or Claude SDK agent, or any other framework via agent_tool. Generates auth config, connector initialization, and tool_utils- or agent_tool-decorated tool functions. Use when adding a connector to an agent or setting up a new agent with a connector.
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. "
),
)
():
connector.execute(entity, action, params {})
"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.
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 capabilitiesasyncdefstripe_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.
Reference demo: connector-sdk/examples/demo_agent.py (mocked, no credentials needed: --mock).
Unsupported Frameworks (agent_tool)
For frameworks without a native tool_utils strategy (anything other than pydantic-ai, LangChain, OpenAI Agents, FastMCP), use agent_tool — the progressive-docs sibling of tool_utils. Decorate three functions; the role is inferred from each signature ((entity, action, ...) → execute, (section, ...) → docs, () → inspect; extra params allowed, ambiguous signatures must pass the role explicitly, e.g. agent_tool("execute")):
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.
