Airbyte Agents MCP works great for Claude Code, but you may need it for something more programmatic, something that runs deep within your codebase and needs to run without human intervention. Or maybe you need an agent harness to handle complicated tasks with subagents.
Airbyte Agents can offer some solid tools for this use case. Airbyte's seamless connection to tools allows your agent to build a business context and act based on that data. You can find more about this process in Airbyte's documentation on Connect, Ask, Act .
This tutorial walks you through how to connect Claude Agent SDK to Airbyte Agents SDK with the use case of a bug triage agent.
Build a bug triage agent Given a prompt like “triage today's new GitHub issues,” this agent will read the issues opened in the last day through a GitHub connector, cross-reference Linear to catch duplicates, group what it finds by severity, and post a short summary to your #engineering Slack channel. By the end you'll have a runnable Python script and a clear picture of how to connect Airbyte Agents with Claude Agent SDK.
You can also use the Claude Agent SDK example in Airbyte's quickstarts repository.
Prerequisites While you don't need prior Claude Agent SDK experience, you will need the following to get started:
Python 3.11 or newer. The Claude Agent SDK bundles the Claude Code CLI, so there's nothing else to install for it to run. An ANTHROPIC_API_KEY. The Agent SDK uses it to run Claude. An Airbyte account. You'll need a client ID, a client secret, and the name of your Airbyte workspace. The Free tier offers 1,000 Agent Operations and a daily Context Store refresh, which is enough to follow along. uv or pip installed for dependency management. This tutorial defaults to uv. An understanding of async Python. If you haven't worked with async or need a refresher, check out Python's conceptual overview . Step 1: Connect your data to Airbyte In your Airbyte workspace, connect the following sources:
Airbyte Agents ships 50+ agent connectors with more on the way, so you can include any supported tool for this workflow. Confirm a connector is live in your workspace from Python with list_connectors().
While you're in the workspace, copy your client ID , client secret , and workspace name . You'll need all three in Step 2.
After this step, three systems are authenticated into your Context Store, ready to query.
Step 2: Set up the project and your .env Create a project and add the two SDKs plus python-dotenv:
uv init bug-triage-agent
cd bug-triage-agent
uv add airbyte-agent-sdk claude-agent-sdk python-dotenvIf you aren't using uv, pip3 install airbyte-agent-sdk claude-agent-sdk python-dotenv gets you the same packages.
Put your credentials in a .env file next to the script:
.env
ANTHROPIC_API_KEY =your_anthropic_key
AIRBYTE_CLIENT_ID =your_client_id
AIRBYTE_CLIENT_SECRET =your_client_secret
AIRBYTE_WORKSPACE_NAME =your_workspace_nameLoad it at the top of your script. connect() reads these variables from the environment, so you don't build or pass an auth object:
from dotenv import load_dotenv
load_dotenv()Your project should now have both SDKs installed and your credentials loaded from .env.
Step 3: Connect Airbyte to the Claude Agent SDK as a tool In the Claude Agent SDK, a custom tool is a Python function wrapped with the @tool decorator, then bundled into an in-process MCP server with create_sdk_mcp_server. The @tool decorator takes three things: a name, a description to tell Claude when to call it, and an input schema.
Airbyte gives you three callables per connector: inspect_connector, read_skill_docs, and execute. For major agent frameworks, Airbyte bundles the three callables into one call: build_connector_tools(connector). Rather than load a connector's whole catalog into the prompt up front, the agent introspects the connector just in time: it inspects the connector, reads the skill docs for the operation it's about to run, then executes.
While build_connector_tools can register those callables straight into some frameworks, the Claude Agent SDK isn't one of them, which means you must wrap the three callables yourself. build_connector_tools(connector).as_list() hands you the plain async callables, and this small factory turns each connector's three callables into three Claude Agent SDK tools:
import json
from claude_agent_sdk import tool
from airbyte_agent_sdk import build_connector_tools
TOOL_SCHEMAS = {
"inspect_connector" : {"type" : "object" , "properties" : {}},
"read_skill_docs" : {
"type" : "object" ,
"properties" : {
"section" : {
"type" : "string" ,
"description" : "Section id from the outline. Omit to get the outline itself." ,
}
},
},
"execute" : {
"type" : "object" ,
"properties" : {
"entity" : {"type" : "string" , "description" : "Entity to operate on, e.g. 'issues'" },
"action" : {
"type" : "string" ,
"description" : "One of: list, get, create, update, delete, api_search" ,
},
"params" : {
"type" : "object" ,
"description" : "Parameters for the action (filters, body fields, ids)" ,
"additionalProperties" : True ,
},
},
"required" : ["entity" , "action" ],
},
}
def make_airbyte_tools (slug, connector ):
callables = {fn.__name__: fn for fn in build_connector_tools(connector).as_list()}
def wrap (name ):
underlying = callables[name]
@tool(f"{slug} _{name} " , underlying.__doc__ or name, TOOL_SCHEMAS[name]
async def _run (args ):
try :
result = await underlying(**args)
except Exception as exc:
return {"content" : [{"type" : "text" , "text" : str (exc)}], "is_error" : True }
return {"content" : [{"type" : "text" , "text" : json.dumps(result, default=str )}]}
return _run
return [wrap(name) for name in ("inspect_connector" , "read_skill_docs" , "execute" )]Here's how the three calls work together, and why the order matters:
inspect_connector() reports the connector's metadata, confirms whether its Context Store is ready, and resolves the skill-doc id the other two calls use. For more information on how Context Store allows for easier searching and filtering, see Airbyte documentation .read_skill_docs() with no section returns an outline of the connector's entities and actions. read_skill_docs(section="...") drills into the exact guidance for the operation the agent is about to run.execute(entity, action, params) runs the operation and returns a structured result with data (the records) and meta (pagination cursors).This progressive flow keeps the agent's context small: it reads only the one section it needs before executing, instead of loading every entity and action up front. Airbyte generates both the skill docs and the connector SDK from the same connector definition, so they stay accurate as the connector grows.
Two details worth flagging:
The handler returns a content array of blocks, which is the shape the Claude Agent SDK expects back from every tool; Claude reads the text block as the tool result. The handler catches exceptions and returns is_error: True. When the agent guesses a wrong read_skill_docs section, the error message carries the valid outline, so Claude can read it and retry instead of the run aborting. Point the factory at any connector and you get its three tools. The agent inspects, reads the relevant docs, and executes all on its own.
After this step, you can turn any connected Airbyte system into a tool for Claude to call.
Step 4: Give the agent a task and run it Now assemble the agent. Construct the three connectors (each reads your credentials from the environment), run a health check on each so a bad connection surfaces before you debug agent logic, build the three tools for each, and register them all in a single in-process MCP server:
Because these calls are async, they live inside an async def main() that you kick off with asyncio.run(main()):
import asyncio
from airbyte_agent_sdk import connect
from claude_agent_sdk import (
create_sdk_mcp_server,
query,
ClaudeAgentOptions,
AssistantMessage,
TextBlock,
ResultMessage,
)
async def main ():
connectors = []
for slug in ("github" , "linear" , "slack" ):
connectors.append((slug, connect(slug)))
for name, connector in connectors:
health = await connector.check()
print (f"{name} : {health.status} " )
tools = []
for slug, connector in connectors:
tools.extend(make_airbyte_tools(slug, connector))
server = create_sdk_mcp_server(name="airbyte" , version="1.0.0" , tools=tools)The connector.check() call is your friend. If a tool comes back anything other than healthy, you have an auth or workspace problem, not an agent problem, and you know that before Claude runs.
Then, still inside main(), run the agent.
Pass the server through ClaudeAgentOptions. Pre-approve your three tools in allowed_tools so they run without a permission prompt. Give Claude a job and stream the result. The asyncio.run(main()) line at the bottom (outside main()) is what actually starts the agent.
options = ClaudeAgentOptions(
mcp_servers={"airbyte" : server},
allowed_tools=["mcp__airbyte__*" ],
system_prompt=(
"You triage engineering bugs. For each connector, inspect it and read its "
"skill docs to learn the entities and actions before you execute. Check "
"Linear for duplicates before creating anything, and keep summaries short."
),
)
prompt = (
"Look at the GitHub issues opened in the last 24 hours. Check Linear for "
"related or duplicate issues, group what you find by severity, and post a "
"short triage summary to the #engineering Slack channel."
)
async for message in query(prompt=prompt, options=options):
if isinstance (message, AssistantMessage):
for block in message.content:
if isinstance (block, TextBlock):
print (block.text)
elif isinstance (message, ResultMessage) and message.subtype == "success" :
print (message.result)
asyncio.run(main())Note the wildcard in allowed_tools. An SDK MCP tool is addressed as mcp__{server_name}__{tool_name}, so mcp__airbyte__* pre-approves all three tools (inspect, read docs, execute) for every connector on the airbyte server. Without it, each call hits a permission prompt.
After this step, run the script to produce a triage summary posted to Slack assembled from live GitHub and Linear data. The full, runnable program is in airbyte-claude-agent-sdk-example.py. Drop your .env next to it, install the dependencies, and run the script.
You can use the following uv command, assuming your file has the same name as the example script:
uv run airbyte-claude-agent-sdk-example.pyTakeaways With Airbyte, you no longer need to connect each system by hand. You can offer the tools through one unified system, let Airbyte maintain the authorization, and ask the agent to make inferences and take actions based on your data. This is Airbyte's Connect, Ask, Act model: connect once, the agent asks across systems, and it acts by writing back through the same interface.
From here, you can:
Tighten the triage loop : have the agent create a Linear issue automatically when a GitHub bug has no matching ticket, then reply on the GitHub issue with the Linear link, so nothing gets triaged twice.Build a scheduled workflow : post a morning digest of overnight issues to Slack before standup so the team walks in to a triaged list instead of an inbox organized by timestamp.Add support context : connect Salesforce or Zendesk to link bugs to support tickets. When the bug is resolved, update the relevant tickets so support can close the loop.This is just one example. Any time you need your agent to access multiple sources to build robust context, use Airbyte Agents to help build better context windows and reduce token usage. For more information about how Airbyte Agents can save tokens, see our benchmarks .
Airbyte