airbyte-agent-connectors

# Airbyte Agent SDK Type-safe connector execution framework with blessed connectors and full IDE autocomplete. ## Overview The Airbyte Agent SDK gives AI agents access to 50+ third-party APIs through strongly typed, well-documented tools. Connectors can run through the Airbyte platform (which manages credentials, rate limiting, and execution) or locally in OSS mode. ## How to install ```bash uv pip install airbyte-agent-sdk ``` ## Documentation Full documentation is available at [docs.airbyte.com/ai-agents/about/](https://docs.airbyte.com/ai-agents/about/). ## Tool integration The SDK ships a hosted tool builder and two decorators for turning connector calls into LLM tools with retry-aware exception translation, output-size guards, and framework-specific error signalling. - **`build_connector_tools(connector, framework="...")`** — preferred for hosted connector agents. Returns `inspect_connector`, `read_skill_docs`, and `execute` callables bound to one connector. Hosted connectors, or local connectors passed an explicit `docs_provider`, use outline-only guidance and tell the agent to inspect/read docs before execution; local/offline connectors without a docs provider keep generated YAML-derived rich docs. Pass `use_progressive_docs=False` to make `tools.as_list()` expose only `execute` with the legacy rich description. - **`@<Connector>.tool_utils`** — preferred for typed connectors. Auto-detects the installed framework (pydantic-ai, LangChain, OpenAI Agents, or FastMCP) and composes [`translate_exceptions`](https://airbytehq.github.io/airbyte-embedded/airbyte_agent_sdk/translation.html) under the hood. Pass `framework="..."` to override auto-detection. Forwards `update_docstring`, `max_output_chars`, `framework`, `internal_retries`, `should_internal_retry`, and `exhausted_runtime_failure_message`. - **`@translate_exceptions`** — same translation behaviour for any callable that is not a generated `Connector` (custom helpers, eval harnesses, ad-hoc tools). The builder and decorators preserve async callables, `__name__`, and `__doc__`. Transient runtime failures (429/5xx, network, timeout) can be retried silently via `internal_retries=N`. Output exceeding `max_output_chars` (default 100 KB) is converted to the framework's retry signal so the LLM can narrow the query. > **Pick one decorator per tool.** Stacking `@translate_exceptions` over `@<Connector>.tool_utils` (or vice versa) is detected at decoration time: the inner layer is preserved and the outer layer logs a warning and short-circuits, so double-translation is impossible. ### Hosted connector tools ```python from pydantic_ai import Agent from airbyte_agent_sdk import build_connector_tools from airbyte_agent_sdk.connectors.stripe import StripeConnector from airbyte_agent_sdk.types import AirbyteAuthConfig stripe = StripeConnector( auth_config=AirbyteAuthConfig( airbyte_client_id="client_abc123", airbyte_client_secret="secret_xyz789", connector_id="src_123", ) ) tools = build_connector_tools(stripe, framework="pydantic_ai") agent = Agent("openai:gpt-4o", tools=tools.as_list()) ``` The model-facing docs flow is `inspect_connector()` -> `read_skill_docs()` -> `read_skill_docs(section="...")` -> `execute(...)`. The docs tool binds the hosted `docs_skill_id` internally, so the model only passes an optional `section`. To opt out of the progressive inspect/docs flow: ```python tools = build_connector_tools(stripe, framework="pydantic_ai", use_progressive_docs=False) agent = Agent("openai:gpt-4o", tools=tools.as_list()) # exposes execute only ``` ### pydantic-ai ```python from pydantic_ai import Agent from airbyte_agent_sdk.connectors.stripe import StripeConnector agent = Agent("openai:gpt-4o") @agent.tool_plain @StripeConnector.tool_utils async def list_customers(limit: int = 10) -> list[dict]: async with StripeConnector(connector_id="src_123") as stripe: result = await stripe.execute("customers", "list", params={"limit": limit}) return result.data ``` Failures raise `pydantic_ai.ModelRetry` so the agent can retry with corrected arguments. ### LangChain ```python from langchain_core.tools import StructuredTool from airbyte_agent_sdk.connectors.stripe import StripeConnector @StripeConnector.tool_utils(framework="langchain") async def list_customers(limit: int = 10) -> list[dict]: async with StripeConnector(connector_id="src_123") as stripe: result = await stripe.execute("customers", "list", params={"limit": limit}) return result.data tool = StructuredTool.from_function( coroutine=list_customers, name="list_customers", description="List Stripe customers.", handle_tool_error=True, # surfaces ToolException as the tool's string result ) ``` Failures raise `langchain_core.tools.ToolException`; `handle_tool_error=True` turns that into the tool's string result for the LLM. > Alternative for non-typed callables: replace `@StripeConnector.tool_utils(framework="langchain")` with `@translate_exceptions(framework="langchain")` from `airbyte_agent_sdk`. ### OpenAI Agents ```python from agents import Agent, function_tool from airbyte_agent_sdk.connectors.stripe import StripeConnector @function_tool @StripeConnector.tool_utils(framework="openai_agents") async def list_customers(limit: int = 10) -> list[dict]: async with StripeConnector(connector_id="src_123") as stripe: result = await stripe.execute("customers", "list", params={"limit": limit}) return result.data agent = Agent(name="stripe", tools=[list_customers]) ``` Note: the OpenAI Agents strategy uses **catch-and-return-string semantics** — `tool_utils` catches the failure and *returns* a string (e.g. `"ConnectorValidationError: entity must be one of: ..."`) instead of raising. The OpenAI runner serialises this string verbatim into the tool result the LLM sees. > Alternative for non-typed callables: replace `@StripeConnector.tool_utils(framework="openai_agents")` with `@translate_exceptions(framework="openai_agents")` from `airbyte_agent_sdk`. ### FastMCP ```python from fastmcp import FastMCP from airbyte_agent_sdk.connectors.stripe import StripeConnector mcp = FastMCP("stripe-tools") @mcp.tool() @StripeConnector.tool_utils(framework="mcp") async def list_customers(limit: int = 10) -> list[dict]: async with StripeConnector(connector_id="src_123") as stripe: result = await stripe.execute("customers", "list", params={"limit": limit}) return result.data ``` Failures raise `fastmcp.exceptions.ToolError`, which FastMCP serialises as an MCP error response to the client. See the [`translate_exceptions`](https://airbytehq.github.io/airbyte-embedded/airbyte_agent_sdk/translation.html) reference for advanced kwargs (`internal_retries`, `should_internal_retry`, `exhausted_runtime_failure_message`). ## How to install the skills The repo ships skills that walk agents through setting up and using the connectors. Three install paths: **skills.sh** (works for Claude Code, Codex, Cursor, OpenCode, and 40+ other agents): ```bash npx skills add airbytehq/airbyte-agent-sdk ``` **Claude Code** (native plugin): ``` /plugin marketplace add airbytehq/airbyte-agent-sdk /plugin install airbyte-agent-sdk@airbyte-agent-sdk ``` **Codex** (clone + symlink): ```bash git clone https://github.com/airbytehq/airbyte-agent-sdk ~/.codex/skills/airbyte-agent-sdk-src ln -s ~/.codex/skills/airbyte-agent-sdk-src/connector-sdk/.claude/skills/* ~/.codex/skills/ ``` See [docs.airbyte.com/ai-agents/about/](https://docs.airbyte.com/ai-agents/about/) for full documentation.