Entity-Action API

Entity-Action API

Airbyte Agent Connectors use a unified entity-action model for interacting with third-party APIs. This document explains the core concepts, available actions, and how to work with results.

Core Concepts

Entity-Action Model

Every connector operation follows the pattern:

result = await connector.execute(entity, action, params)

• Entity: The resource type you're working with (e.g., customers, issues, contacts)
• Action: The operation to perform (e.g., list, get, create, update)
• Params: Parameters specific to the entity and action

This model provides a consistent interface across all 21 connectors, regardless of the underlying API's design.

Example

# Stripe: List customers
await connector.execute("customers", "list", {"limit": 10})

# GitHub: Get a specific issue
await connector.execute("issues", "get", {
    "owner": "airbytehq",
    "repo": "airbyte",
    "number": 123
})

# Salesforce: Search opportunities
await connector.execute("opportunities", "api_search", {
    "query": "Amount > 10000"
})

Available Actions

get

Retrieve a single record by its identifier.

# Get a specific customer
result = await connector.execute("customers", "get", {"id": "cus_xxx"})

# Get a specific repository
result = await connector.execute("repositories", "get", {
    "owner": "airbytehq",
    "repo": "airbyte"
})

Common Parameters:

• id - Primary identifier for the record
• Additional identifiers vary by entity (e.g., owner/repo for GitHub)
• fields - Optional array of field names to return

list

Retrieve multiple records with optional filtering and pagination.

# List customers
result = await connector.execute("customers", "list", {
    "limit": 100,
    "starting_after": "cus_xxx"  # Pagination cursor
})

# List issues with filters
result = await connector.execute("issues", "list", {
    "owner": "airbytehq",
    "repo": "airbyte",
    "states": ["OPEN"],
    "per_page": 50
})

Common Parameters:

• limit or per_page - Number of records to return
• after or starting_after - Cursor for pagination
• Filter parameters vary by entity (e.g., states, created_at)
• fields - Optional array of field names to return

create

Create a new record.

# Create a customer
result = await connector.execute("customers", "create", {
    "email": "user@example.com",
    "name": "Jane Doe",
    "metadata": {"source": "agent"}
})

# Create a product
result = await connector.execute("products", "create", {
    "name": "Premium Plan",
    "description": "Full access subscription"
})

Returns: The created record with its assigned ID.

update

Modify an existing record.

# Update a customer
result = await connector.execute("customers", "update", {
    "id": "cus_xxx",
    "name": "Jane Smith",
    "metadata": {"updated": "true"}
})

# Update a product
result = await connector.execute("products", "update", {
    "id": "prod_xxx",
    "name": "Premium Plan v2"
})

Returns: The updated record.

delete

Remove a record.

# Delete a customer
result = await connector.execute("customers", "delete", {"id": "cus_xxx"})

# Delete a product
result = await connector.execute("products", "delete", {"id": "prod_xxx"})

Returns: Confirmation of deletion.

api_search

Search using the API's native search syntax. This provides access to powerful server-side filtering.

# GitHub: Search repositories
result = await connector.execute("repositories", "api_search", {
    "query": "language:python stars:>1000 topic:machine-learning"
})

# Stripe: Search customers
result = await connector.execute("customers", "api_search", {
    "query": "email:'user@example.com' AND metadata['status']:'active'"
})

# Salesforce: SOQL query
result = await connector.execute("accounts", "api_search", {
    "query": "Industry = 'Technology' AND AnnualRevenue > 1000000"
})

Note: Query syntax varies by connector. Refer to each connector's REFERENCE.md for syntax details.

download

Download file content (available on connectors with file entities).

# Google Drive: Download file
result = await connector.execute("files", "download", {
    "file_id": "1abc123..."
})

# The result.data will be an AsyncIterator[bytes] for streaming

Execution Result

All operations return an ExecutionResult:

@dataclass
class ExecutionResult:
    success: bool
    data: dict[str, Any] | list[dict] | AsyncIterator[bytes]
    error: str | None = None
    meta: dict[str, Any] | None = None

Handling Results

result = await connector.execute("customers", "list", {"limit": 10})

if result.success:
    # Access the data
    customers = result.data
    for customer in customers:
        print(f"{customer['id']}: {customer['email']}")

    # Check pagination metadata
    if result.meta and result.meta.get("has_more"):
        next_cursor = result.meta.get("next_cursor")
        print(f"More results available, cursor: {next_cursor}")
else:
    # Handle error
    print(f"Operation failed: {result.error}")

Result Structure by Action

Pagination

Cursor-Based Pagination

Most connectors use cursor-based pagination:

async def list_all_customers(connector):
    """Iterate through all customers with pagination."""
    all_customers = []
    cursor = None

    while True:
        params = {"limit": 100}
        if cursor:
            params["starting_after"] = cursor

        result = await connector.execute("customers", "list", params)

        if not result.success:
            raise Exception(f"Failed: {result.error}")

        all_customers.extend(result.data)

        # Check if more pages exist
        if result.meta and result.meta.get("has_more"):
            cursor = result.meta.get("next_cursor")
        else:
            break

    return all_customers

Pagination Parameters by Connector

Field Selection

Many connectors support selecting specific fields to reduce response size:

# Only return id, email, and name fields
result = await connector.execute("customers", "list", {
    "limit": 100,
    "fields": ["id", "email", "name"]
})

# GitHub: Select specific repository fields
result = await connector.execute("repositories", "get", {
    "owner": "airbytehq",
    "repo": "airbyte",
    "fields": ["name", "description", "stargazerCount", "forkCount"]
})

Validation

Validate operations before executing them to catch errors early. This is especially useful in agent workflows where you want to verify parameters before making API calls.

Basic Validation

# Validate that entity, action, and params are correct
validation = await connector.validate_operation("customers", "get", {"id": "cus_xxx"})

if validation.is_valid:
    result = await connector.execute("customers", "get", {"id": "cus_xxx"})
else:
    print(f"Validation errors: {validation.errors}")

Validation in Agent Tools

@agent.tool_plain
async def safe_execute(entity: str, action: str, params: dict | None = None) -> str:
    """Execute an operation with validation."""
    params = params or {}

    # Validate first
    validation = await connector.validate_operation(entity, action, params)
    if not validation.is_valid:
        return f"Invalid operation: {', '.join(validation.errors)}"

    # Execute if valid
    result = await connector.execute(entity, action, params)
    if result.success:
        return json.dumps(result.data)
    else:
        return f"Error: {result.error}"

What Validation Checks

Validation Response

@dataclass
class ValidationResult:
    is_valid: bool
    errors: list[str]  # List of validation error messages
    warnings: list[str]  # Non-blocking issues

Example: Catching Missing Parameters

# Missing required 'owner' parameter
validation = await connector.validate_operation("repositories", "get", {
    "repo": "airbyte"  # Missing 'owner'
})

print(validation.is_valid)  # False
print(validation.errors)    # ["Missing required parameter: owner"]

Example: Invalid Parameter Value

# Invalid state value
validation = await connector.validate_operation("issues", "list", {
    "owner": "airbytehq",
    "repo": "airbyte",
    "states": ["INVALID_STATE"]  # Should be OPEN, CLOSED, or MERGED
})

print(validation.is_valid)  # False
print(validation.errors)    # ["Invalid value for 'states': INVALID_STATE"]

Error Types

The connectors raise specific errors for common failure modes:

Error Handling Example

from airbyte_agent_core.errors import (
    EntityNotFoundError,
    ActionNotSupportedError,
    MissingParameterError,
    AuthenticationError,
    RateLimitError
)

try:
    result = await connector.execute("customers", "get", {"id": "cus_xxx"})
except EntityNotFoundError:
    print("'customers' is not a valid entity for this connector")
except ActionNotSupportedError:
    print("'get' action not supported for customers")
except MissingParameterError as e:
    print(f"Missing required parameter: {e}")
except AuthenticationError:
    print("Authentication failed - check your credentials")
except RateLimitError:
    print("Rate limited - wait and retry")

Using with Agent Frameworks

Generic Tool Pattern

Create a single tool that handles any entity/action combination:

@agent.tool_plain
@Connector.tool_utils
async def execute(entity: str, action: str, params: dict | None = None) -> str:
    """Execute an operation on the connector.

    Args:
        entity: The resource type (e.g., 'customers', 'issues')
        action: The operation (e.g., 'list', 'get', 'create')
        params: Parameters for the operation

    Returns:
        JSON string of the result
    """
    result = await connector.execute(entity, action, params or {})
    if result.success:
        return json.dumps(result.data)
    else:
        return f"Error: {result.error}"

Entity-Specific Tools

Create focused tools for common operations:

@agent.tool_plain
async def list_customers(limit: int = 10, email_filter: str | None = None) -> str:
    """List Stripe customers."""
    params = {"limit": limit}
    if email_filter:
        params["email"] = email_filter
    result = await connector.execute("customers", "list", params)
    return json.dumps(result.data) if result.success else f"Error: {result.error}"

@agent.tool_plain
async def get_customer(customer_id: str) -> str:
    """Get a specific Stripe customer by ID."""
    result = await connector.execute("customers", "get", {"id": customer_id})
    return json.dumps(result.data) if result.success else f"Error: {result.error}"

Discovering Available Entities and Actions

Programmatic Discovery

# List available entities
entities = connector.list_entities()
print(entities)  # ['customers', 'invoices', 'charges', ...]

# Get entity details
schema = connector.describe_entity("customers")
print(schema)  # Shows available actions and parameters

Reference Documentation

Each connector's REFERENCE.md provides a complete listing:

• GitHub REFERENCE.md
• Stripe REFERENCE.md
• Salesforce REFERENCE.md

The reference docs include:

• All entities with descriptions
• Available actions per entity
• Required and optional parameters
• Example code for each operation