Twilio

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

The Twilio agent connector is a Python package that equips AI agents to interact with Twilio 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 Twilio REST API. Provides read and write access to core Twilio resources including accounts, calls, messages, recordings, conferences, incoming phone numbers, usage records, addresses, queues, transcriptions, and outgoing caller IDs. Write operations include sending SMS/MMS messages, placing outbound calls, and provisioning phone numbers. Uses HTTP Basic authentication with Account SID and Auth Token.

Example prompts

The Twilio connector is optimized to handle prompts like these.

• List all calls from the last 7 days
• Show me recent inbound SMS messages
• List all active phone numbers on my account
• Show me details for a specific call
• List all recordings
• Show me conference calls
• List usage records for my account
• Show me all queues
• List outgoing caller IDs
• Show me addresses on my account
• List transcriptions
• Send an SMS message to +15558675310 saying 'Hello from Twilio!'
• Place an outbound call to +15558675310 with the message 'Your appointment is confirmed'
• Provision a new phone number with area code 415
• Send a WhatsApp message to +15558675310
• Send an MMS with an image to +15558675310
• What are my top 10 most expensive calls this month?
• How many SMS messages did I send vs receive in the last 30 days?
• Summarize my usage costs by category
• Which phone numbers have the most incoming calls?
• Show me all failed messages and their error codes
• What is the average call duration for outbound calls?

Unsupported prompts

The Twilio connector isn't currently able to handle prompts like these.

• Delete a recording
• Delete a phone number
• Delete a message
• Create a new queue

Entities and actions

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

Entity	Actions
Accounts	List, Get, Context Store Search
Calls	List, Create, Get, Context Store Search
Messages	List, Create, Get, Context Store Search
Incoming Phone Numbers	List, Create, Get, Context Store Search
Recordings	List, Get, Context Store Search
Conferences	List, Get, Context Store Search
Usage Records	List, Context Store Search
Addresses	List, Get, Context Store Search
Queues	List, Get, Context Store Search
Transcriptions	List, Get, Context Store Search
Outgoing Caller Ids	List, Get, Context Store Search

Twilio API docs

See the official Twilio 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 TwilioConnector and reads AIRBYTE_CLIENT_ID / AIRBYTE_CLIENT_SECRET from the environment:

Pydantic AI

Pydantic AI

from pydantic_ai import Agent
from airbyte_agent_sdk import connect
from airbyte_agent_sdk.connectors.twilio import TwilioConnector

connector = connect("twilio", workspace_name="<your_workspace_name>")

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

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

LangChain

LangChain

from langchain_core.tools import tool
from airbyte_agent_sdk import connect
from airbyte_agent_sdk.connectors.twilio import TwilioConnector

connector = connect("twilio", workspace_name="<your_workspace_name>")

@tool
@TwilioConnector.tool_utils
async def twilio_execute(entity: str, action: str, params: dict | None = None):
    """Execute Twilio 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

OpenAI Agents

from agents import Agent, function_tool
from airbyte_agent_sdk import connect
from airbyte_agent_sdk.connectors.twilio import TwilioConnector

connector = connect("twilio", 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)
@TwilioConnector.tool_utils(framework="openai_agents")
async def twilio_execute(entity: str, action: str, params: dict | None = None):
    """Execute Twilio 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="Twilio Assistant", tools=[twilio_execute])

FastMCP

FastMCP

from fastmcp import FastMCP
from airbyte_agent_sdk import connect
from airbyte_agent_sdk.connectors.twilio import TwilioConnector

connector = connect("twilio", workspace_name="<your_workspace_name>")

mcp = FastMCP("Twilio Agent")

@mcp.tool
@TwilioConnector.tool_utils
async def twilio_execute(entity: str, action: str, params: dict | None = None):
    """Execute Twilio 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

Pydantic AI

from pydantic_ai import Agent
from airbyte_agent_sdk.connectors.twilio import TwilioConnector
from airbyte_agent_sdk.types import AirbyteAuthConfig

connector = TwilioConnector(
    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
@TwilioConnector.tool_utils
async def twilio_execute(entity: str, action: str, params: dict | None = None):
    return await connector.execute(entity, action, params or {})

LangChain

LangChain

from langchain_core.tools import tool
from airbyte_agent_sdk.connectors.twilio import TwilioConnector
from airbyte_agent_sdk.types import AirbyteAuthConfig

connector = TwilioConnector(
    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
@TwilioConnector.tool_utils
async def twilio_execute(entity: str, action: str, params: dict | None = None):
    """Execute Twilio 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

OpenAI Agents

from agents import Agent, function_tool
from airbyte_agent_sdk.connectors.twilio import TwilioConnector
from airbyte_agent_sdk.types import AirbyteAuthConfig

connector = TwilioConnector(
    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)
@TwilioConnector.tool_utils(framework="openai_agents")
async def twilio_execute(entity: str, action: str, params: dict | None = None):
    """Execute Twilio 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="Twilio Assistant", tools=[twilio_execute])

FastMCP

FastMCP

from fastmcp import FastMCP
from airbyte_agent_sdk.connectors.twilio import TwilioConnector
from airbyte_agent_sdk.types import AirbyteAuthConfig

connector = TwilioConnector(
    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("Twilio Agent")

@mcp.tool
@TwilioConnector.tool_utils
async def twilio_execute(entity: str, action: str, params: dict | None = None):
    """Execute Twilio 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

Pydantic AI

from pydantic_ai import Agent
from airbyte_agent_sdk.connectors.twilio import TwilioConnector
from airbyte_agent_sdk.connectors.twilio.models import TwilioAuthConfig

connector = TwilioConnector(
    auth_config=TwilioAuthConfig(
        account_sid="<Your Twilio Account SID (starts with AC)>",
        auth_token="<Your Twilio Auth Token>"
    )
)

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

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

LangChain

LangChain

from langchain_core.tools import tool
from airbyte_agent_sdk.connectors.twilio import TwilioConnector
from airbyte_agent_sdk.connectors.twilio.models import TwilioAuthConfig

connector = TwilioConnector(
    auth_config=TwilioAuthConfig(
        account_sid="<Your Twilio Account SID (starts with AC)>",
        auth_token="<Your Twilio Auth Token>"
    )
)

@tool
@TwilioConnector.tool_utils
async def twilio_execute(entity: str, action: str, params: dict | None = None):
    """Execute Twilio 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

OpenAI Agents

from agents import Agent, function_tool
from airbyte_agent_sdk.connectors.twilio import TwilioConnector
from airbyte_agent_sdk.connectors.twilio.models import TwilioAuthConfig

connector = TwilioConnector(
    auth_config=TwilioAuthConfig(
        account_sid="<Your Twilio Account SID (starts with AC)>",
        auth_token="<Your Twilio Auth Token>"
    )
)

# strict_mode=False because `params: dict` is permissive and the default strict
# JSON schema rejects objects with additionalProperties.
@function_tool(strict_mode=False)
@TwilioConnector.tool_utils(framework="openai_agents")
async def twilio_execute(entity: str, action: str, params: dict | None = None):
    """Execute Twilio 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="Twilio Assistant", tools=[twilio_execute])

FastMCP

FastMCP

from fastmcp import FastMCP
from airbyte_agent_sdk.connectors.twilio import TwilioConnector
from airbyte_agent_sdk.connectors.twilio.models import TwilioAuthConfig

connector = TwilioConnector(
    auth_config=TwilioAuthConfig(
        account_sid="<Your Twilio Account SID (starts with AC)>",
        auth_token="<Your Twilio Auth Token>"
    )
)

mcp = FastMCP("Twilio Agent")

@mcp.tool
@TwilioConnector.tool_utils
async def twilio_execute(entity: str, action: str, params: dict | None = None):
    """Execute Twilio 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.4