bootstrapping-agent

用途

bootstrapping-agent 是一个用于“智能体接入 Airbyte 连接器”的引导技能，面向使用 PydanticAI 或 Claude SDK 构建智能体的开发者。它的目标是把 Airbyte 的各类数据连接器（如 Stripe、Zendesk Support、HubSpot 等）接入到智能体中，使其能够作为工具被调用。

主要能力

生成认证配置与连接器初始化代码（采用托管模式）。
生成带 tool_utils 装饰的工具函数。
根据开发者已有导入自动识别框架：
- from pydantic_ai import Agent → 使用 PydanticAI 模式。
- from anthropic import Anthropic → 使用 Claude SDK 模式。
- 无法判断时，询问开发者使用哪个框架。
处理连接器命名约定：连接器 slug 中的连字符在子模块路径中转换为下划线，例如 zendesk-support 对应 airbyte_agent_sdk.connectors.zendesk_support。

使用方式

安装 SDK：

uv pip install airbyte-agent-sdk

单个 airbyte-agent-sdk 包包含所有类型化连接器，从 airbyte_agent_sdk.connectors.{slug} 导入。tool_utils、list_entities() 和 entity_schema() 仅对类型化连接器可用。

PydanticAI 核心模式示例：

import os
from pydantic_ai import Agent
from airbyte_agent_sdk import AirbyteAuthConfig
from airbyte_agent_sdk.connectors.stripe import StripeConnector

connector = StripeConnector(
    auth_config=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"),
    )
)

agent = Agent(
    "provider:model",
    system_prompt=(
        "You are a helpful assistant with access to Stripe. "
        "Use the stripe_execute tool to look up customer, invoice, and balance data. "
        "Ask for clarification if a request is ambiguous."
    ),
)

@agent.tool_plain
@StripeConnector.tool_utils
async def stripe_execute(entity: str, action: str, params: dict | None = None):
    return await connector.execute(entity, action, params or {})

始终使用托管模式：使用 AirbyteAuthConfig，配合 airbyte_client_id 与 airbyte_client_secret。不要生成本地认证代码。

装饰器叠加顺序：框架装饰器在上，tool_utils 在下。tool_utils 是 @classmethod，应使用 StripeConnector.tool_utils，而不是 connector.tool_utils。

@agent.tool_plain           # 框架将其注册为工具
@StripeConnector.tool_utils # 用连接器能力丰富 docstring
async def stripe_execute(...):

自动重试转换：tool_utils 会自动将可重试错误转换为框架的重试信号（PydanticAI 中为 ModelRetry），无需额外的装饰器。

参考演示：connector-sdk/examples/demo_agent.py（已 mock，无需凭据：--mock）。

校验配置：

check = await connector.check()
if check.status == "healthy":
    print(f"Connected — checked {check.checked_entity}/{check.checked_action}")
else:
    print(f"Failed: {check.error}")

连接器导入与类名对应示例：

Connector	Import	Class
stripe	airbyte_agent_sdk.connectors.stripe	StripeConnector
zendesk-support	airbyte_agent_sdk.connectors.zendesk_support	ZendeskSupportConnector
hubspot	airbyte_agent_sdk.connectors.hubspot	HubspotConnector

环境变量（需放入 .env）：

AIRBYTE_CLIENT_ID=your_client_id
AIRBYTE_CLIENT_SECRET=your_client_secret
AIRBYTE_WORKSPACE_NAME=your_workspace_name

相关参考：SDK API 参考（完整 API 签名与选项）、PydanticAI 模式（完整可运行示例）、Claude SDK 模式（Anthropic Python SDK 示例）。

适用场景

需要让 PydanticAI 或 Claude SDK 智能体通过 Airbyte 连接器访问外部数据源（如 Stripe、Zendesk Support、HubSpot）时，可用该技能完成 SDK 安装、认证配置、连接器初始化与工具函数编写。

限制

仅支持类型化连接器的 tool_utils、list_entities()、entity_schema()。
只支持托管模式，不适用本地认证代码。
未提供收费、价格、平台支持范围等信息。

技能介绍

用途

主要能力

使用方式

适用场景

限制