SuperAPI AI Directory
Partner with us中文AI newsFavorites
Skillsbuilding-multi-connector-agent

building-multi-connector-agent

Agents & models

View repository

Updated 10-15 16:45

About this skill

Building a Multi-Connector Agent

Use this when an agent needs two or more Airbyte connectors.

Going from Single to Multi

The bootstrapping-agent skill shows the default single-connector pattern: AirbyteAuthConfig plus build_connector_tools. The builder's execute, inspect_connector, and read_skill_docs names are fixed, so the tool sets for more than one connector collide when registered on the same agent. Renaming the callables at registration avoids the collision, but the generated execute guidance still names inspect_connector and read_skill_docs, pointing the model at the wrong tools. Multi-connector agents write their own tool functions with @Connector.agent_tool(...), which weaves the real names in through inspect_tool= and docs_tool=:

  1. Build a single AirbyteAuthConfig(...) so credentials are shared across connectors.
  2. Construct one typed connector per service (e.g. JiraConnector(auth_config=auth)).
  3. Define execute, inspect, and docs functions for each connector, each with its own @Connector.agent_tool(...) decorator.

Use connector-specific function names so the framework registers six unambiguous tools for two connectors.

Install the SDK

uv pip install airbyte-agent-sdk

The single airbyte-agent-sdk package bundles every typed connector without per-connector installs.

Core Pattern (PydanticAI)

import os
from pydantic_ai import Agent
from airbyte_agent_sdk import AirbyteAuthConfig
from airbyte_agent_sdk.connectors.jira import JiraConnector
from airbyte_agent_sdk.connectors.slack import SlackConnector

# Shared credentials — all connectors reuse the same AirbyteAuthConfig
auth = AirbyteAuthConfig(
    airbyte_client_id=os.getenv("AIRBYTE_CLIENT_ID"),
    airbyte_client_secret=os.getenv("AIRBYTE_CLIENT_SECRET"),
    workspace_name=os.getenv("AIRBYTE_WORKSPACE_NAME", "default"),
)

jira = JiraConnector(auth_config=auth)
slack = SlackConnector(auth_config=auth)

If the workspace contains multiple connectors of the same type, pin one by passing connector_id=os.getenv("JIRA_CONNECTOR_ID") to the constructor.

Three Progressive Tools Per Connector

Each connector gets its own execute, inspect, and docs functions. Do not combine connectors into a mega-tool; separate names give the LLM clear, independent tool descriptions and let it read only the relevant connector docs.

agent_tool is a @classmethod — decorate with @JiraConnector.agent_tool(...), not @jira.agent_tool(...).

agent = Agent(
    "<provider:model>",
    system_prompt=(
        "You are a helpful assistant with access to Jira and Slack. "
        "Use the jira_execute tool to read Jira issues and the slack_execute tool to post messages. "
        "Ask for clarification if a request is ambiguous."
    ),
)

@agent.tool_plain
@JiraConnector.agent_tool(
    framework="pydantic_ai",
    inspect_tool="jira_inspect",
    docs_tool="jira_read_docs",
)
async def jira_execute(entity: str, action: str, params: dict | None = None):
    return await jira.execute(entity, action, params or {})

@agent.tool_plain
@JiraConnector.agent_tool(framework="pydantic_ai")
async def jira_inspect():
    return await jira.inspect_connector()

@agent.tool_plain
@JiraConnector.agent_tool(framework="pydantic_ai")
async def jira_read_docs(section: str | None = None):
    return await jira.read_skill_docs(section)

@agent.tool_plain
@SlackConnector.agent_tool(
    framework="pydantic_ai",
    inspect_tool="slack_inspect",
    docs_tool="slack_read_docs",
)
async def slack_execute(entity: str, action: str, params: dict | None = None):
    return await slack.execute(entity, action, params or {})

@agent.tool_plain
@SlackConnector.agent_tool(framework="pydantic_ai")
async def slack_inspect():
    return await slack.inspect_connector()

@agent.tool_plain
@SlackConnector.agent_tool(framework="pydantic_ai")
async def slack_read_docs(section: str | None = None):
    return await slack.read_skill_docs(section)

For LangChain, OpenAI Agents, or FastMCP, keep the same three-functions-per-connector structure and change framework= plus the outer registration decorator. Omit framework= for unsupported frameworks and handle AirbyteToolError in the dispatch loop.

Connector.tool_utils is deprecated and remains available only for backwards compatibility with existing one-tool integrations. Do not generate it for new multi-connector agents.

System Prompt

Describe the agent's purpose and what each connector does:

agent = Agent(
    "<provider:model>",
    system_prompt=(
        "You are a customer support assistant. "
        "Use the stripe tool to look up customer billing data. "
        "Use the jira tool to create and track support tickets. "
        "Use the slack tool to notify the support team."
    ),
)

Run Loop

PydanticAI

import asyncio

async def main():
    result = await agent.run("Find open P0 bugs and post a summary to #engineering")
    print(result.output)
    await jira.close()
    await slack.close()

asyncio.run(main())

Claude SDK (Anthropic Python)

See Claude SDK patterns for the full message loop with tool handling.

Project Structure

For a new agent project:

my-agent/
├── pyproject.toml       # dependencies: airbyte-agent-sdk, pydantic-ai or anthropic
├── .env                 # AIRBYTE_CLIENT_ID, AIRBYTE_CLIENT_SECRET, AIRBYTE_WORKSPACE_NAME
├── agent.py             # Entry point: auth config, connectors, agent + tools, run loop
└── README.md

pyproject.toml

[project]
name = "my-agent"
requires-python = ">=3.11"
dependencies = [
    "airbyte-agent-sdk",
    "pydantic-ai",
    "python-dotenv",
]

Environment Variables

AIRBYTE_CLIENT_ID=your_client_id
AIRBYTE_CLIENT_SECRET=your_client_secret
AIRBYTE_WORKSPACE_NAME=your_workspace_name

References