airbyte-agent-sdk — independently scanned and version-tracked by SaferSkills.
SaferSkills independently audited airbyte-agent-sdk (Plugin) and scored it 100/100 (green). The audit ran 55 deterministic rules across Security, Supply Chain, Maintenance, Transparency, and Community; it found 0 high-severity and 0 lower-severity findings. The full rule-by-rule trace and per-finding evidence are below. Free, methodology-open.
Findings & checks · 0 flagged
Every scanned point with the score it earned and what moved between them.
First recorded scan — no prior version to compare against.
The primary manifest — the file an agent reads to learn what this artifact does.
Type-safe connector execution framework with blessed connectors and full IDE autocomplete.
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.
uv pip install airbyte-agent-sdkFull documentation is available at docs.airbyte.com/ai-agents/about/.
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.
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.translate_exceptions 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.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_exceptionsover@<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.
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:
tools = build_connector_tools(stripe, framework="pydantic_ai", use_progressive_docs=False)
agent = Agent("openai:gpt-4o", tools=tools.as_list()) # exposes execute onlyfrom 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.dataFailures raise pydantic_ai.ModelRetry so the agent can retry with corrected arguments.
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")fromairbyte_agent_sdk.
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")fromairbyte_agent_sdk.
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.dataFailures raise fastmcp.exceptions.ToolError, which FastMCP serialises as an MCP error response to the client.
See the translate_exceptions reference for advanced kwargs (internal_retries, should_internal_retry, exhausted_runtime_failure_message).
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):
npx skills add airbytehq/airbyte-agent-sdkClaude Code (native plugin):
/plugin marketplace add airbytehq/airbyte-agent-sdk
/plugin install airbyte-agent-sdk@airbyte-agent-sdkCodex (clone + symlink):
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/ for full documentation.
~30 seconds. Free. No account. Every finding cites a rule and a line of evidence.