The SDK for production-grade agents.

auth = AirbyteAuthConfig(
   airbyte_client_id=os.environ["AIRBYTE_CLIENT_ID"],
   airbyte_client_secret=os.environ["AIRBYTE_CLIENT_SECRET"]
)

hubspot = HubspotConnector(auth_config=auth)
stripe = StripeConnector(auth_config=auth)

email = "customer@example.com"

# Search for the contact in HubSpot
contact = (await hubspot.contacts.context_store_search(
   query={"filter": {"eq": {"properties.email": email}}},
   limit=1
)).data[0]

# Search for the customer in Stripe
customer = (await stripe.customers.list(email=email, limit=1)).data[0]

# List subscriptions for that Stripe customer
subs = await stripe.subscriptions.list(customer=customer.id)

connector = LinearConnector(
 auth_config=AirbyteAuthConfig(    airbyte_client_id=os.getenv("AIRBYTE_CLIENT_ID"),
    airbyte_client_secret=os.getenv("AIRBYTE_CLIENT_SECRET),
   )
)

# Updated the model name to gpt-5.5
‍agent = Agent("openai:gpt-5.5", system_prompt="You have access to Linear.")

@agent.tool_plain  # assumes you're using Pydantic AI
@LinearConnector.tool_utils
‍async def linear_execute(
   entity: str,
   action: str,
   params: dict | None = None,
):
   return await connector.issues.create(
       team_id="<your_team_id>",
       title="My new issue",
       description="Created via the Airbyte Agent SDK",
   )