Customer-Io

This connector is optimized for AI agents. For the data replication connector, see Customer Io.

The Customer-Io agent connector is a Python package that equips AI agents to interact with Customer-Io through strongly typed, well-documented tools. It's ready to use directly in your Python app, in an agent framework, or exposed through an MCP.

Connector for the Customer.io App API, providing read access to campaigns, newsletters, segments, messages, activities, sender identities, snippets, collections, reporting webhooks, exports, and transactional message templates. Customer.io is a messaging platform that lets you send automated emails, push notifications, SMS, and in-app messages based on customer behavior. This connector retrieves data from the App API (not the Track API) and requires a Bearer-type App API key generated in your Customer.io workspace settings.

Example prompts

The Customer-Io connector is optimized to handle prompts like these.

List all campaigns in Customer.io
Show me all newsletters
What segments are defined in my workspace?
Get the details of campaign 42
List all sender identities
Show me all reporting webhooks
What snippets do we have?
List all collections
Show recent activities
Create a snippet called 'footer' with content '<p>Thanks!</p>'
Update the snippet 'header' to say 'Welcome back!'
Create a new collection called 'products'
Create a reporting webhook for email events
Create a manual segment called 'VIP Customers'
Export all customers matching a segment
Send a transactional email to [email protected]
Send an SMS notification to +15551234567
Send a push notification to user 123
Trigger broadcast campaign 42
List all transactional message templates
Get the details of transactional message 5
Show the content variants of transactional template 3
Update the subject of transactional content 139 in template 3
Which campaigns are currently active?
Find newsletters sent in the last month
What are the most recent email deliveries?
Which exports have completed?

Unsupported prompts

The Customer-Io connector isn't currently able to handle prompts like these.

Create a new campaign
Delete a segment
Delete a snippet
Delete a collection
Delete a reporting webhook
Delete a newsletter
Send a newsletter via API
Schedule a newsletter via API

Entities and actions

This connector supports the following entities and actions. For more details, see this connector's full reference documentation.

Entity	Actions
Campaigns	List, Get, Context Store Search
Campaign Actions	List, Get, Context Store Search
Newsletters	List, Get, Context Store Search
Segments	List, Create, Get
Messages	List, Get
Activities	List
Sender Identities	List, Get
Snippets	List, Create, Update
Collections	List, Create, Get, Update
Reporting Webhooks	List, Create, Get, Update
Exports	List, Create, Get
Transactional Messages	List, Get
Transactional Message Contents	List, Update
Transactional Email	Create
Transactional Sms	Create
Transactional Push	Create
Transactional Inbox Message	Create
Broadcast Trigger	Create

Customer-Io API docs

See the official Customer-Io API reference.

SDK installation

uv pip install airbyte-agent-sdk

SDK usage

Connectors can run in hosted or open source mode.

Hosted

In hosted mode, API credentials are stored securely in Airbyte Agents. You provide your Airbyte credentials instead. If your Airbyte client can access multiple organizations, also set organization_id.

This example assumes you've already authenticated your connector with Airbyte. See Authentication to learn more about authenticating. If you need a step-by-step guide, see the hosted execution tutorial.

The connect() factory returns a fully typed CustomerIoConnector and reads AIRBYTE_CLIENT_ID / AIRBYTE_CLIENT_SECRET from the environment:

Pydantic AI
LangChain
OpenAI Agents
FastMCP

Pydantic AI
from pydantic_ai import Agent
from airbyte_agent_sdk import connect
from airbyte_agent_sdk.connectors.customer_io import CustomerIoConnector

connector = connect("customer-io", workspace_name="<your_workspace_name>")

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

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

LangChain
from langchain_core.tools import tool
from airbyte_agent_sdk import connect
from airbyte_agent_sdk.connectors.customer_io import CustomerIoConnector

connector = connect("customer-io", workspace_name="<your_workspace_name>")

@tool
@CustomerIoConnector.tool_utils
async def customer_io_execute(entity: str, action: str, params: dict | None = None):
    """Execute Customer-Io connector operations."""
    result = await connector.execute(entity, action, params or {})
    # connector.execute returns a Pydantic envelope for typed actions; fall back to raw data otherwise.
    return result.model_dump(mode="json") if hasattr(result, "model_dump") else result

OpenAI Agents
from agents import Agent, function_tool
from airbyte_agent_sdk import connect
from airbyte_agent_sdk.connectors.customer_io import CustomerIoConnector

connector = connect("customer-io", workspace_name="<your_workspace_name>")

# strict_mode=False because `params: dict` is permissive and the default strict
# JSON schema rejects objects with additionalProperties.
@function_tool(strict_mode=False)
@CustomerIoConnector.tool_utils(framework="openai_agents")
async def customer_io_execute(entity: str, action: str, params: dict | None = None):
    """Execute Customer-Io connector operations."""
    result = await connector.execute(entity, action, params or {})
    return result.model_dump(mode="json") if hasattr(result, "model_dump") else result

agent = Agent(name="Customer-Io Assistant", tools=[customer_io_execute])

FastMCP
from fastmcp import FastMCP
from airbyte_agent_sdk import connect
from airbyte_agent_sdk.connectors.customer_io import CustomerIoConnector

connector = connect("customer-io", workspace_name="<your_workspace_name>")

mcp = FastMCP("Customer-Io Agent")

@mcp.tool
@CustomerIoConnector.tool_utils
async def customer_io_execute(entity: str, action: str, params: dict | None = None):
    """Execute Customer-Io connector operations."""
    result = await connector.execute(entity, action, params or {})
    return result.model_dump(mode="json") if hasattr(result, "model_dump") else result

Or pass credentials explicitly (equivalent, useful when you're not loading them from the environment):

Pydantic AI
LangChain
OpenAI Agents
FastMCP

Pydantic AI
from pydantic_ai import Agent
from airbyte_agent_sdk.connectors.customer_io import CustomerIoConnector
from airbyte_agent_sdk.types import AirbyteAuthConfig

connector = CustomerIoConnector(
    auth_config=AirbyteAuthConfig(
        workspace_name="<your_workspace_name>",
        organization_id="<your_organization_id>",  # Optional for multi-org clients
        airbyte_client_id="<your-client-id>",
        airbyte_client_secret="<your-client-secret>"
    )
)

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

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

LangChain
from langchain_core.tools import tool
from airbyte_agent_sdk.connectors.customer_io import CustomerIoConnector
from airbyte_agent_sdk.types import AirbyteAuthConfig

connector = CustomerIoConnector(
    auth_config=AirbyteAuthConfig(
        workspace_name="<your_workspace_name>",
        organization_id="<your_organization_id>",  # Optional for multi-org clients
        airbyte_client_id="<your-client-id>",
        airbyte_client_secret="<your-client-secret>"
    )
)

@tool
@CustomerIoConnector.tool_utils
async def customer_io_execute(entity: str, action: str, params: dict | None = None):
    """Execute Customer-Io connector operations."""
    result = await connector.execute(entity, action, params or {})
    # connector.execute returns a Pydantic envelope for typed actions; fall back to raw data otherwise.
    return result.model_dump(mode="json") if hasattr(result, "model_dump") else result

OpenAI Agents
from agents import Agent, function_tool
from airbyte_agent_sdk.connectors.customer_io import CustomerIoConnector
from airbyte_agent_sdk.types import AirbyteAuthConfig

connector = CustomerIoConnector(
    auth_config=AirbyteAuthConfig(
        workspace_name="<your_workspace_name>",
        organization_id="<your_organization_id>",  # Optional for multi-org clients
        airbyte_client_id="<your-client-id>",
        airbyte_client_secret="<your-client-secret>"
    )
)

# strict_mode=False because `params: dict` is permissive and the default strict
# JSON schema rejects objects with additionalProperties.
@function_tool(strict_mode=False)
@CustomerIoConnector.tool_utils(framework="openai_agents")
async def customer_io_execute(entity: str, action: str, params: dict | None = None):
    """Execute Customer-Io connector operations."""
    result = await connector.execute(entity, action, params or {})
    return result.model_dump(mode="json") if hasattr(result, "model_dump") else result

agent = Agent(name="Customer-Io Assistant", tools=[customer_io_execute])

FastMCP
from fastmcp import FastMCP
from airbyte_agent_sdk.connectors.customer_io import CustomerIoConnector
from airbyte_agent_sdk.types import AirbyteAuthConfig

connector = CustomerIoConnector(
    auth_config=AirbyteAuthConfig(
        workspace_name="<your_workspace_name>",
        organization_id="<your_organization_id>",  # Optional for multi-org clients
        airbyte_client_id="<your-client-id>",
        airbyte_client_secret="<your-client-secret>"
    )
)

mcp = FastMCP("Customer-Io Agent")

@mcp.tool
@CustomerIoConnector.tool_utils
async def customer_io_execute(entity: str, action: str, params: dict | None = None):
    """Execute Customer-Io connector operations."""
    result = await connector.execute(entity, action, params or {})
    return result.model_dump(mode="json") if hasattr(result, "model_dump") else result

Open source

In open source mode, you provide API credentials directly to the connector.

Pydantic AI
LangChain
OpenAI Agents
FastMCP

Pydantic AI
from pydantic_ai import Agent
from airbyte_agent_sdk.connectors.customer_io import CustomerIoConnector
from airbyte_agent_sdk.connectors.customer_io.models import CustomerIoAuthConfig

connector = CustomerIoConnector(
    auth_config=CustomerIoAuthConfig(
        app_api_key="<Your Customer.io App API key. Generate one in your workspace settings at Settings > API Credentials > App API Key.
>"
    )
)

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

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

LangChain
from langchain_core.tools import tool
from airbyte_agent_sdk.connectors.customer_io import CustomerIoConnector
from airbyte_agent_sdk.connectors.customer_io.models import CustomerIoAuthConfig

connector = CustomerIoConnector(
    auth_config=CustomerIoAuthConfig(
        app_api_key="<Your Customer.io App API key. Generate one in your workspace settings at Settings > API Credentials > App API Key.
>"
    )
)

@tool
@CustomerIoConnector.tool_utils
async def customer_io_execute(entity: str, action: str, params: dict | None = None):
    """Execute Customer-Io connector operations."""
    result = await connector.execute(entity, action, params or {})
    # connector.execute returns a Pydantic envelope for typed actions; fall back to raw data otherwise.
    return result.model_dump(mode="json") if hasattr(result, "model_dump") else result

OpenAI Agents
from agents import Agent, function_tool
from airbyte_agent_sdk.connectors.customer_io import CustomerIoConnector
from airbyte_agent_sdk.connectors.customer_io.models import CustomerIoAuthConfig

connector = CustomerIoConnector(
    auth_config=CustomerIoAuthConfig(
        app_api_key="<Your Customer.io App API key. Generate one in your workspace settings at Settings > API Credentials > App API Key.
>"
    )
)

# strict_mode=False because `params: dict` is permissive and the default strict
# JSON schema rejects objects with additionalProperties.
@function_tool(strict_mode=False)
@CustomerIoConnector.tool_utils(framework="openai_agents")
async def customer_io_execute(entity: str, action: str, params: dict | None = None):
    """Execute Customer-Io connector operations."""
    result = await connector.execute(entity, action, params or {})
    return result.model_dump(mode="json") if hasattr(result, "model_dump") else result

agent = Agent(name="Customer-Io Assistant", tools=[customer_io_execute])

FastMCP
from fastmcp import FastMCP
from airbyte_agent_sdk.connectors.customer_io import CustomerIoConnector
from airbyte_agent_sdk.connectors.customer_io.models import CustomerIoAuthConfig

connector = CustomerIoConnector(
    auth_config=CustomerIoAuthConfig(
        app_api_key="<Your Customer.io App API key. Generate one in your workspace settings at Settings > API Credentials > App API Key.
>"
    )
)

mcp = FastMCP("Customer-Io Agent")

@mcp.tool
@CustomerIoConnector.tool_utils
async def customer_io_execute(entity: str, action: str, params: dict | None = None):
    """Execute Customer-Io connector operations."""
    result = await connector.execute(entity, action, params or {})
    return result.model_dump(mode="json") if hasattr(result, "model_dump") else result

Authentication

For all authentication options, see the connector's authentication documentation.

Version information

Connector version: 1.0.0

Example prompts​

Unsupported prompts​

Entities and actions​

Customer-Io API docs​

SDK installation​

SDK usage​

Hosted​

Open source​

Authentication​

Version information​