Content
# 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.
Connection Info
You Might Also Like
markitdown
MarkItDown-MCP is a lightweight server for converting URIs to Markdown.
markitdown
Python tool for converting files and office documents to Markdown.
firecrawl
Firecrawl MCP Server enables web scraping, crawling, and content extraction.
Filesystem
Node.js MCP Server for filesystem operations with dynamic access control.
TrendRadar
TrendRadar: Your hotspot assistant for real news in just 30 seconds.
mempalace
The highest-scoring AI memory system ever benchmarked. And it's free.