The Airbyte Agent SDK provides a framework for AI agents to access external systems through reliable, permission-aware tools. This Python-based SDK gives developers the ability to integrate their AI systems with 50+ third-party APIs through strongly typed connectors that run either through the Airbyte platform or locally in open-source mode.
What it does The Airbyte Agent SDK focuses on making it easier for AI agents to interact with external systems. Key features include:
- Type-safe connector execution framework with strongly typed, well-documented tools
- Access to 50+ third-party APIs
- Support for running through Airbyte platform or locally in OSS mode
- Exception translation for transient runtime failures
- Output-size guards to prevent large responses
- Framework-specific error signaling
Getting it running Installation is straightforward using uv:
uv pip install airbyte-agent-sdk
The SDK provides two main decorators for turning connector calls into LLM tools:
@<Connector>.tool_utils- Preferred for typed connectors@translate_exceptions- For any callable not generated from a Connector
These decorators handle retry-aware exception translation, output-size guards, and framework-specific error signaling. They preserve sync/async functionality, __name__, and __doc__ attributes.
Framework Integration The SDK supports multiple AI frameworks with specific examples provided:
For pydantic-ai:
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
For LangChain:
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,
)
For OpenAI Agents:
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])
Who this is for The Airbyte Agent SDK is designed for developers building AI applications that need to interact with external systems. It's particularly valuable for those working with multiple APIs who want type safety, proper error handling, and integration with major AI frameworks. The SDK handles complexities like rate limiting, credential management, and transient failures, allowing developers to focus on their AI application logic rather than API integration details.
How it compares The SDK distinguishes itself through its type safety, comprehensive error handling, and framework-agnostic approach. Unlike many API integration tools, it's specifically designed for AI agent use cases, with features like automatic retry logic and framework-specific error signaling. The SDK currently has 119 GitHub stars and is actively maintained by the Airbyte team.
The Airbyte Agent SDK may be overkill for simple API integration needs but excels for complex AI agent applications requiring robust interaction with multiple external systems. Full documentation is available at https://docs.airbyte.com/ai-agents/about/.
Comments