Skip to main content

Execute operations

AvailableFreeAvailableIndividualAvailableTeamAvailableCustom Compare plans

Once you've added a connector to your workspace, you can run operations against it from Python. The SDK offers direct execution and patterns for exposing a connector as a tool to an AI agent framework.

Direct execution​

The connect() factory takes a connector slug and returns an execution object. Call its execute(entity, action, params) method to run an operation. When your workspace has exactly one connector of a given type, you don't need to pass a connector_id. The SDK resolves the connector by its slug automatically.

agent.py
import asyncio
from airbyte_agent_sdk import connect

async def main():
    github = connect("github")
    try:
        result = await github.execute("issues", "list", params={"per_page": 10})
        for row in result.data:
            print(row)
    finally:
        await github.close()

asyncio.run(main())
  • entity is the resource, such as issues, repositories, or pull_requests.
  • action is one of the connector's supported actions, such as list or get. Some connectors support additional actions like context_store_search, api_search, or download; check the connector's reference page.
  • params contains action-specific arguments. The exact keys are connector- and entity-specific. GitHub's issues.list accepts per_page, for example, while other connectors paginate via cursor. Use list_entities() to discover the parameters a connector supports at runtime.
  • Always wrap the call in try/finally and await connector.close() once you're done to release the underlying HTTP client.

See the connector's page in the Connectors reference for the entities and actions it supports.

Typed connectors and HostedExecutor​

For connectors with a generated typed submodule, connect() returns a typed connector with IDE autocompletion, method-level docstrings, and structured call shortcuts. For example: await hubspot.contacts.list(limit=10). The agent connectors page lists every connector; the Slug column is the string to pass to connect().

For every other connector in the bundled registry, connect() returns a generic HostedExecutor with the same execute(entity, action, params) method but without typed shortcuts. The execution behavior is otherwise identical.

connect() raises ValueError if the slug isn't in the bundled registry (the message lists every supported slug) or if no Airbyte credentials are available. It does not raise when a typed submodule is missing. YAML-only connectors return a HostedExecutor.

Multiple connectors of the same type​

If your workspace has more than one connector of a given type (for example, two separate Stripe accounts), slug resolution is ambiguous. Pass an explicit connector_id to connect() so the SDK knows which one to target:

agent.py
stripe_us = connect("stripe", connector_id="<us_account_connector_id>")
stripe_eu = connect("stripe", connector_id="<eu_account_connector_id>")

For patterns that look up a connector ID without hard-coding it, see Get a connector.

Expose a connector as an agent tool​

When you wrap a connector as a tool for an AI agent, the agent needs to know which entities and actions exist, what parameters each takes, and how pagination works. The recommended pattern is build_connector_tools, which binds ready-to-use tools that let the agent read just-in-time skill docs before it executes.

The SDK offers these options. Pick the first one that fits:

  1. build_connector_tools: the simplest, preferred default on a supported framework (Pydantic AI, LangChain, OpenAI Agents SDK, FastMCP) when you don't need to write the tool bodies yourself. It binds one connector's tool set per agent.
  2. agent_tool: use it when you need custom tool bodies, a framework the SDK doesn't natively support, or an agent that uses more than one connector.
  3. tool_utils: deprecated. Keep it only in existing integrations; don't use it for new tools.
  4. translate_exceptions: the same exception translation for any callable that isn't a generated connector tool, such as a custom helper or evaluation harness. agent_tool and tool_utils already include this translation, so don't stack translate_exceptions on top of them. If you do, the SDK detects the double wrap, logs a warning, and leaves the existing wrapper in place.

build_connector_tools(connector) returns a ConnectorTools object with three callables bound to one connector: inspect_connector, read_skill_docs, and execute. Pass framework= so runtime errors surface as that framework's retry signal, then register all three with tools.as_list().

Instead of packing the whole connector schema into one static tool description, the agent works through a progressive introspection flow:

inspect_connector() -> read_skill_docs() -> read_skill_docs(section="...") -> execute(entity, action, params)
  • inspect_connector() returns the connector's hosted metadata and Context Store readiness, and resolves the skill-doc ID the other tools use.
  • read_skill_docs() with no section returns the outline and general guidance. read_skill_docs(section="<id>") returns the exact entity and action guidance the agent needs before it executes. The section ID must be copied verbatim from the outline, including its prefix (actions.issues.list, not issues.list); anything else returns an error that the agent has to recover from.
  • execute(entity, action, params) runs the operation.

The SDK binds the skill-doc ID internally, so the model only passes an optional section. This keeps the agent's context small: it reads the outline, drills into the one section it needs, then executes, instead of loading every entity and action up front. Skill docs are served by Airbyte from the same connector definition the SDK is generated from, so they stay in sync with the connector.

connect("github") returns a typed connector when a generated submodule exists, but build_connector_tools also accepts a generic HostedExecutor for YAML-only connectors. Either works with the same three tools.

note

Skill docs are hosted by Airbyte and served by the platform. If you point the SDK at a connector running in open source mode (no hosted backend), build_connector_tools still returns the same three tools, but inspect_connector reports "mode": "local" and read_skill_docs falls back to the connector's generated (YAML-derived) description instead of hosted section docs. execute still runs directly against the connector.

To expose only execute with a single generated description instead of the progressive flow, pass use_progressive_docs=False. tools.as_list() then returns just the execute tool.

The builder names its tools inspect_connector, read_skill_docs, and execute, and the SDK can't rename them, so the builder binds one connector's tool set per agent. If you register two connectors' tool sets with one agent, the names collide. Renaming the callables yourself at registration avoids the collision, but in the progressive flow the generated execute guidance still tells the agent to call inspect_connector and read_skill_docs, so it points at the wrong tools. For an agent that uses more than one connector, use agent_tool with connector-specific function names and pass those names through inspect_tool= and docs_tool=.

Register the tools with your framework​

tools.as_list() returns plain async callables, so you can register them with any framework. Set framework= to match the one you use.

Pass tools.as_list() straight to the agent:

agent.py
from pydantic_ai import Agent
from airbyte_agent_sdk import build_connector_tools, connect

github = connect("github")
tools = build_connector_tools(github, framework="pydantic_ai")

agent = Agent("openai:gpt-4o", tools=tools.as_list())

Custom tool bodies with agent_tool​

build_connector_tools writes the tool bodies for you. When you need your own to log calls, post-process results, restrict reachable agent operations, run a framework the SDK doesn't natively support, or give one agent more than one connector, use the generated connector's agent_tool decorator instead. Use it even on a supported framework when you need custom tool bodies, and pass framework= so failures surface as that framework's signal. It keeps the same progressive flow: you write the three functions, and the decorator attaches guidance that steers the agent from inspect to docs to execute.

Decorate one function per role, and register all three. The framework's registration decorator (for example, @agent.tool_plain) goes on top and agent_tool goes underneath. agent_tool is a class method: call GithubConnector.agent_tool(...), not github.agent_tool(...). The role is inferred from the signature, and extra parameters are allowed:

  • (entity, action, ...) is execute
  • (section, ...) is docs
  • () is inspect

Pass the role explicitly (such as agent_tool("execute")) if a wrapper has an ambiguous signature. The optional inspect_tool= and docs_tool= names are woven into the execute guidance so the agent refers to your registered tool names rather than generic ones.

agent.py
from pydantic_ai import Agent
from airbyte_agent_sdk import connect
from airbyte_agent_sdk.connectors.github import GithubConnector

agent = Agent("openai:gpt-4o")
github = connect("github")

@agent.tool_plain
@GithubConnector.agent_tool(
    framework="pydantic_ai",
    inspect_tool="github_inspect",
    docs_tool="github_read_docs",
)
async def github_execute(entity: str, action: str, params: dict | None = None):
    return await github.execute(entity, action, params or {})

@agent.tool_plain
@GithubConnector.agent_tool(framework="pydantic_ai")
async def github_inspect():
    return await github.inspect_connector()

@agent.tool_plain
@GithubConnector.agent_tool(framework="pydantic_ai")
async def github_read_docs(section: str | None = None):
    return await github.read_skill_docs(section)

For multi-connector agents, add the connector name to the start of each function name to avoid naming ambiguity, and pass the inspect and docs names to the execute decorator's inspect_tool= and docs_tool= so the guidance names the right sibling tools. Give each connector its own:

  • <connector>_execute
  • <connector>_inspect
  • <connector>_read_docs

Choose a failure signal with framework=​

framework= controls what a tool failure looks like to the agent. It behaves the same way on build_connector_tools, agent_tool, and translate_exceptions.

framework=A tool failure surfaces as
"pydantic_ai"Raises pydantic_ai.ModelRetry, so the agent retries.
"langchain"Raises langchain_core.tools.ToolException. LangChain aborts the run unless you feed the message back to the model: pass handle_tool_error=True when you construct the tool, or follow Surface tool errors back to the model.
"openai_agents"Returns the failure message as the tool result instead of raising, which is what the OpenAI Agents SDK expects. Register the tool with function_tool(..., strict_mode=False) so the open-ended params dict is accepted.
"mcp"Raises fastmcp.exceptions.ToolError, which FastMCP serializes as a failed tool result.
"none"Raises AirbyteToolError.

build_connector_tools, tool_utils, and translate_exceptions auto-detect an installed framework when you omit framework=, and fall back to "none" with a warning if they find none. agent_tool never auto-detects: it defaults to "none".

If you pass a framework= whose package isn't installed, the SDK raises RuntimeError when it translates a tool failure, not when you decorate the function. Install the framework package or omit framework=.

Unsupported frameworks and raw LLM loops​

On a framework the SDK doesn't cover, or in a hand-rolled dispatch loop against a model API, omit framework= and handle AirbyteToolError yourself. Advertise each function to the model using its docstring as the tool description, and return the error message as the tool result so the model can correct itself.

agent.py
from airbyte_agent_sdk import AirbyteToolError

tools = {fn.__name__: fn for fn in (github_inspect, github_read_docs, github_execute)}

# tool_name and tool_args come from the model's tool call, so don't assume the name exists.
handler = tools.get(tool_name)
if handler is None:
    tool_result = f"Unknown tool: {tool_name}"
else:
    try:
        tool_result = await handler(**tool_args)
    except AirbyteToolError as err:
        tool_result = str(err)

AirbyteToolError inherits from AirbyteError and keeps the original exception on __cause__.

Other patterns​

The patterns below predate build_connector_tools. They bind a single execute tool with the connector's full catalog baked into the description up front, rather than letting the agent read skill docs on demand. Prefer build_connector_tools, or agent_tool when you need your own tool bodies.

Manual docstrings​

Define one tool per operation with a hand-written docstring. Use this when you want to expose a narrow set of operations or need full control over parameters.

agent.py
from pydantic_ai import Agent
from airbyte_agent_sdk import connect

agent = Agent("openai:gpt-4o")
github = connect("github")

@agent.tool_plain
async def list_issues(owner: str, repo: str, limit: int = 10) -> str:
    """List open issues in a GitHub repository."""
    result = await github.issues.list(owner=owner, repo=repo, states=["OPEN"], per_page=limit)
    return str(result.data)

The docstring becomes the tool description the LLM sees. Function parameters become the tool's input schema.

Auto-generated tool descriptions with tool_utils​

tool_utils is deprecated. It remains available so existing integrations keep working and can migrate on their own schedule, and it doesn't warn at runtime, but don't use it for new tools. New agents should use build_connector_tools, or agent_tool when they need custom tool bodies. Like build_connector_tools, it auto-detects an installed framework and includes exception translation, so don't stack translate_exceptions on it.

The decorator replaces the wrapped function's docstring with a generated description that includes every entity, action, required and optional parameter, and response shape. The LLM then sees every operation the connector supports with no extra wiring, and pays for the whole catalog in context on every call, which is what the progressive flow avoids.

agent.py
from pydantic_ai import Agent
from airbyte_agent_sdk import connect
from airbyte_agent_sdk.connectors.github import GithubConnector

agent = Agent("openai:gpt-4o")
github = connect("github")

@agent.tool_plain
@GithubConnector.tool_utils
async def github_execute(entity: str, action: str, params: dict | None = None):
    return await github.execute(entity, action, params or {})
Decorator order matters​

The framework decorator (for example, @agent.tool_plain or FastMCP's @mcp.tool) captures __doc__ at decoration time. @Connector.tool_utils must be the inner decorator so it can rewrite __doc__ before the framework reads it.

agent.py
@agent.tool_plain            # Outer: framework captures __doc__
@GithubConnector.tool_utils  # Inner: rewrites __doc__ first
async def github_execute(entity, action, params=None):
    ...

If you reverse the order, the framework captures the original empty docstring and the LLM loses the generated documentation.

Custom docstrings​

Generated docstrings are almost always the right choice. Override them only when your agent specifically misuses the tool and a custom description fixes it.

warning

Custom docstrings can contradict the connector's actual behavior, and they don't update when the connector adds new actions. Prefer generated docstrings unless you have a specific reason not to.

agent.py
@agent.tool_plain
@GithubConnector.tool_utils
async def github_execute(entity: str, action: str, params: dict | None = None):
    """Execute GitHub operations.

    `entity` must be a simple name such as `issues`, `repositories`, or `pull_requests`.
    `action` must be `list`, `get`, or `context_store_search`.
    Pass owner and repo info in the `params` dict, for example:
    `params={"owner": "airbytehq", "repo": "airbyte"}`.
    """
    return await github.execute(entity, action, params or {})

Download files​

Some connectors support a download action for binary entities like attachments, audio recordings, and documents. Download responses return a byte stream instead of JSON.

Normally, you first list a parent resource to find the file's ID, then download the file. The examples below assume zendesk_support = connect("zendesk-support"). Zendesk Support has a generated typed submodule, so connect() returns a typed connector here. YAML-only connectors would return a HostedExecutor and use the generic execute(entity, action, params) API instead.

agent.py
zendesk_support = connect("zendesk-support")
comments = await zendesk_support.ticket_comments.list(ticket_id="456")
for comment in comments.data:
    for attachment in comment.attachments or []:
        print(f"Attachment: {attachment['id']} - {attachment['file_name']}")

Once you have the attachment ID, stream the file to disk:

agent.py
stream = await zendesk_support.attachments.download(attachment_id="12345")
with open("./downloads/ticket_attachment.pdf", "wb") as f:
    async for chunk in stream:
        f.write(chunk)

Or use download_local() to save a file in one call:

agent.py
file_path = await zendesk_support.attachments.download_local(
    attachment_id="12345",
    path="./downloads/ticket_attachment.pdf",
)

To see which entities support download, check the connector's reference page.

Introspection​

On typed connectors, you can ask at runtime what the connector supports. These methods are not available on HostedExecutor; call connect() with a connector that has a generated typed submodule.

list_entities() returns every entity, its available actions, and the parameters each action accepts.

agent.py
entities = github.list_entities()
for entity in entities:
    print(f"{entity['entity_name']}: {entity['available_actions']}")
    # issues: ['list', 'get']

entity_schema(entity) returns the JSON schema for records of that entity, or None if the connector doesn't ship one for that entity. Always guard the result:

agent.py
schema = github.entity_schema("issues")
if schema is None:
    print("No schema available for issues")
else:
    print(list(schema.get("properties", {}).keys()))

Handle errors​

Most SDK-owned errors inherit from AirbyteError, including HTTPStatusError (non-2xx responses from the API) and AuthenticationError (invalid or expired credentials). The hosted execution path also propagates raw httpx errors unwrapped. Catch both in one place.

agent.py
import httpx
from airbyte_agent_sdk import AirbyteError, connect

stripe = connect("stripe")
try:
    result = await stripe.execute("customers", "list", params={"limit": 10})
except (AirbyteError, httpx.HTTPError) as err:
    print(f"Execution failed: {err!r}")

For the full exception hierarchy, including HTTPStatusError and other SDK-defined subclasses in airbyte_agent_sdk.http.exceptions, see the SDK reference.

Was this page helpful?