### Build and Preview Documentation

Source: https://github.com/airbytehq/pyairbyte/blob/main/docs/CONTRIBUTING.md

Use this command to build the documentation and open it in a web browser. If Poe is not installed, prepend with `uv run`.

```console
poe docs-preview
```

```console
uv run poe docs-preview
```

--------------------------------

### Start MCP Server with STDIO Transport

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

Starts the Airbyte MCP server using the command-line interface with STDIO transport, suitable for AI agents like Claude Desktop.

```bash
airbyte-mcp
```

--------------------------------

### Start MCP Server with HTTP Transport

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

Starts the Airbyte MCP server programmatically using Python, configured for HTTP transport on localhost.

```python
python -c "from airbyte.mcp.server import app; app.run(transport='http', host='127.0.0.1', port=8000)"
```

--------------------------------

### Start MCP Server with SSE Transport

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

Starts the Airbyte MCP server programmatically using Python, configured for SSE (Server-Sent Events) transport on localhost.

```python
python -c "from airbyte.mcp.server import app; app.run(transport='sse', host='127.0.0.1', port=8000)"
```

--------------------------------

### Run Coverage Report

Source: https://github.com/airbytehq/pyairbyte/blob/main/docs/CONTRIBUTING.md

Execute this command to generate an HTML coverage report. If Poe is not installed, prepend with `uv run`.

```console
uv run poe coverage-html
```

--------------------------------

### MCP Server Configuration

Source: https://github.com/airbytehq/pyairbyte/blob/main/docs/CONTRIBUTING.md

Example JSON configuration for running the Airbyte MCP server using `uv` as the entrypoint. Ensure paths and environment variables are correctly set for your development environment.

```json
{
  "mcpServers": {
    "airbyte": {
      "command": "uv",
      "args": [
        "--directory=/path/to/repos/PyAirbyte",
        "run",
        "airbyte-mcp"
      ],
      "env": {
        "AIRBYTE_MCP_ENV_FILE": "/path/to/my/.mcp/airbyte_mcp.env",
        "AIRBYTE_CLOUD_MCP_READONLY_MODE": "0",
        "AIRBYTE_CLOUD_MCP_SAFE_MODE": "0"
      }
    }
  }
}
```

--------------------------------

### get_source

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

Instantiates a configured Source object for any Airbyte source connector. It automatically selects the best execution method (YAML manifest, Python pip install, or Docker) unless explicitly overridden and installs the connector if it's missing.

```APIDOC
## get_source — Instantiate a source connector

Returns a configured `Source` object for any Airbyte source connector. Auto-selects the best execution method (YAML manifest, Python pip install, or Docker) unless explicitly overridden. The connector is installed on first use if not already present.

### Parameters

- **connector_name** (string) - Required - The name of the source connector to instantiate (e.g., "source-faker").
- **config** (dict) - Required - The configuration dictionary for the source connector.
- **streams** (list[string], optional) - A list of stream names to be selected for the source.
- **install_if_missing** (bool, optional) - If True, the connector will be installed if it's not already present. Defaults to False.
- **docker_image** (bool, optional) - If True, forces the use of Docker for execution.
- **version** (string, optional) - Specifies a particular version of the connector to use.
- **use_python** (string, optional) - Specifies the Python version to use for the connector's virtual environment.
- **source_manifest** (string, optional) - Path to a local declarative YAML manifest file for a custom source.

### Methods on Source object

- **check()**: Validates the connector configuration.
- **get_available_streams()**: Returns a list of streams available from the connector.
- **config_spec**: JSON Schema describing the connector's configuration fields.
```

--------------------------------

### Test MCP Tools with Poe

Source: https://github.com/airbytehq/pyairbyte/blob/main/docs/CONTRIBUTING.md

Use Poe tasks to test individual MCP tools by providing the tool name and JSON-formatted arguments. Examples include listing connectors, getting specs, validating configs, and running syncs.

```bash
poe mcp-tool-test <tool_name> '<json_args>'

poe mcp-tool-test list_connectors '{}'
poe mcp-tool-test get_config_spec '{"connector_name": "source-pokeapi"}'
poe mcp-tool-test validate_config \
    '{"connector_name": "source-pokeapi", "config": {"pokemon_name": "pikachu"}}'
poe mcp-tool-test run_sync \
    '{"connector_name": "source-pokeapi", "config": {"pokemon_name": "pikachu"}}'

poe mcp-tool-test check_airbyte_cloud_workspace '{}'
poe mcp-tool-test list_deployed_cloud_connections '{}'
```

--------------------------------

### Install Connector with Custom Python Version

Source: https://github.com/airbytehq/pyairbyte/blob/main/README.md

Use the `use_python` argument in `get_source()` to specify a Python version for connector installation, even if it differs from the PyAirbyte's running version. This is useful for compatibility with older connectors.

```python
import airbyte as ab

source = ab.get_source(
    "source-faker",
    use_python="3.10.17",
)
```

--------------------------------

### Claude Desktop MCP Configuration

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

Example JSON configuration for Claude Desktop to connect to an Airbyte MCP server, including environment variables for authentication.

```json
// Claude Desktop MCP config (~/.config/claude/mcp.json)
{
  "mcpServers": {
    "airbyte-mcp": {
      "command": "airbyte-mcp",
      "env": {
        "AIRBYTE_CLOUD_CLIENT_ID": "your-client-id",
        "AIRBYTE_CLOUD_CLIENT_SECRET": "your-client-secret",
        "AIRBYTE_CLOUD_WORKSPACE_ID": "your-workspace-id"
      }
    }
  }
}
```

--------------------------------

### MCP Server (`airbyte-mcp`)

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

PyAirbyte ships a FastMCP server exposing connector management tools to AI agents. Start it with the `airbyte-mcp` CLI command (STDIO transport) or embed it in any MCP-compatible client.

```APIDOC
## MCP Server (`airbyte-mcp`) — AI agent integration

PyAirbyte ships a FastMCP server exposing connector management tools to AI agents. Start it with the `airbyte-mcp` CLI command (STDIO transport) or embed it in any MCP-compatible client.

```bash
# Start with STDIO transport (for Claude Desktop, etc.)
airbyte-mcp

# HTTP transport
python -c "from airbyte.mcp.server import app; app.run(transport='http', host='127.0.0.1', port=8000)"

# SSE transport
python -c "from airbyte.mcp.server import app; app.run(transport='sse', host='127.0.0.1', port=8000)"
```

```json
// Claude Desktop MCP config (~/.config/claude/mcp.json)
{
  "mcpServers": {
    "airbyte-mcp": {
      "command": "airbyte-mcp",
      "env": {
        "AIRBYTE_CLOUD_CLIENT_ID": "your-client-id",
        "AIRBYTE_CLOUD_CLIENT_SECRET": "your-client-secret",
        "AIRBYTE_CLOUD_WORKSPACE_ID": "your-workspace-id"
      }
    }
  }
}
```

Key MCP tools exposed to AI agents:

- **`validate_connector_config`** — Check if a connector config is valid
- **`list_source_streams`** — List streams available in a source
- **`get_source_stream_json_schema`** — Get JSON schema for a specific stream
- **`read_source_stream_records`** — Read up to N records from a source stream
- **`get_stream_previews`** — Get sample records from multiple streams
- **`sync_source_to_cache`** — Run a full sync to local DuckDB cache
- **`list_cached_streams`** — List streams in the default cache
- **`run_sql_query`** — Execute a read-only SQL query against the cache
- **`describe_default_cache`** — Inspect the current default cache
- **`destination_smoke_test`** — Run smoke tests against a destination connector

```python
# Programmatic use of MCP tools (for testing)
from airbyte.mcp.local import (
    validate_connector_config,
    list_source_streams,
    read_source_stream_records,
    sync_source_to_cache,
    run_sql_query,
)

# Validate a config
ok, msg = validate_connector_config(
    connector_name="source-faker",
    config={"count": 100},
)
print(ok, msg)  # True, "Configuration for source-faker is valid!"

# List streams
streams = list_source_streams(
    source_connector_name="source-faker",
    config={"count": 100},
)
print(streams)  # ['products', 'purchases', 'users']

# Read records
records = read_source_stream_records(
    source_connector_name="source-faker",
    config={"count": 50},
    stream_name="users",
    max_records=10,
)
print(records[0])

# Sync and query
sync_source_to_cache(
    source_connector_name="source-faker",
    config={"count": 500},
    streams=["users"],
)
results = run_sql_query("SELECT id, name FROM main.users LIMIT 5")
for row in results:
    print(row)
```
```

--------------------------------

### Get available Airbyte connectors

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

Query the Airbyte connector registry to retrieve a list of all available connector names. Optionally filter results by connector type (source or destination).

```python
import airbyte as ab

# All connectors
all_connectors = ab.get_available_connectors()
print(f"Total connectors: {len(all_connectors)}")

# Filter by source or destination
sources = ab.get_available_connectors(connector_type="source")
destinations = ab.get_available_connectors(connector_type="destination")

```

--------------------------------

### Get Records with Field Normalization

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

Fetches records from a source stream, applying field name normalization (lowercase, special characters removed) and pruning undeclared fields.

```python
records = source.get_records(
    "products",
    normalize_field_names=True,
    prune_undeclared_fields=True,
)
first_50 = list(records)[:50]
```

--------------------------------

### Initialize Caches for Postgres and BigQuery

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

Demonstrates initializing cache objects for PostgreSQL and BigQuery. Ensure necessary credentials and connection details are provided.

```python
pg_cache = PostgresCache(
    host="localhost",
    port=5432,
    database="mydb",
    username="postgres",
    password=ab.get_secret("POSTGRES_PASSWORD"),
    schema_name="airbyte_raw",
)

bq_cache = BigQueryCache(
    project_name="my-gcp-project",
    dataset_name="airbyte_raw",
    credentials_path="/path/to/service_account.json",
)
```

--------------------------------

### Preview Data with get_samples and print_samples

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

Fetches a small number of records from selected streams for quick data preview. `get_samples` returns a dictionary of datasets, while `print_samples` renders formatted tables.

```python
import airbyte as ab

source = ab.get_source("source-faker", config={"count": 100},
                       streams=["users", "products"])

# Get samples as a dict of stream -> InMemoryDataset
samples = source.get_samples(streams=["users", "products"], limit=3)
for stream, dataset in samples.items():
    print(f"{stream}: {list(dataset)}")

# Print rich formatted tables to the terminal
source.print_samples(streams="*", limit=5, on_error="log")
```

--------------------------------

### Generate API Documentation

Source: https://github.com/airbytehq/pyairbyte/blob/main/docs/CONTRIBUTING.md

Run this command to generate API documentation from docstrings. The generated files will be placed in the `docs/generated` folder.

```console
poe docs-generate
```

--------------------------------

### Instantiate an Airbyte Source Connector

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

Use `get_source` to create a configured Source object. The connector is auto-installed if missing. Supports explicit Docker execution, pinned versions, and local manifest files.

```python
import airbyte as ab

# Minimal usage — auto-installs source-faker via uv/pip
source = ab.get_source(
    "source-faker",
    config={"count": 1000},
    streams=["users", "products"],
    install_if_missing=True,
)

# With explicit Docker execution
source_docker = ab.get_source(
    "source-github",
    config={
        "credentials": {
            "personal_access_token": ab.get_secret("GITHUB_PERSONAL_ACCESS_TOKEN"),
        },
        "repositories": ["airbytehq/pyairbyte"],
    },
    docker_image=True,
)

# With a pinned connector version and specific Python version
source_pinned = ab.get_source(
    "source-postgres",
    config={"host": "localhost", "port": 5432, "database": "mydb",
            "username": "user", "password": ab.get_secret("PG_PASSWORD"),
            "ssl_mode": {"mode": "disable"}},
    version="3.6.0",
    use_python="3.10",
)

# From a local declarative YAML manifest
source_manifest = ab.get_source(
    "source-custom",
    config={"api_key": "abc123"},
    source_manifest="/path/to/manifest.yaml",
)

# Validate config and list available streams
source.check()
print(source.get_available_streams())
# ['users', 'products', 'purchases']
print(source.config_spec)  # JSON Schema of all config fields
```

--------------------------------

### Test MCP Server From Source

Source: https://github.com/airbytehq/pyairbyte/blob/main/docs/CONTRIBUTING.md

Synchronize development dependencies and then run the MCP inspect task to test the MCP server from your local source code.

```bash
uv sync --group dev
uv run poe mcp-inspect
```

--------------------------------

### Initialize and use CloudWorkspace

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

Instantiate a CloudWorkspace to manage resources in Airbyte Cloud. Deploy local sources and destinations, list connections, and run syncs.

```python
import airbyte as ab
from airbyte import cloud

workspace = cloud.CloudWorkspace(
    workspace_id=ab.get_secret("AIRBYTE_CLOUD_WORKSPACE_ID"),
    client_id=ab.get_secret("AIRBYTE_CLOUD_CLIENT_ID"),
    client_secret=ab.get_secret("AIRBYTE_CLOUD_CLIENT_SECRET"),
)

# Deploy a local source definition to Cloud
local_source = ab.get_source(
    "source-faker",
    config={\"count\": 5000},
    no_executor=True,   # fetch spec only, no local install needed
)
cloud_source = workspace.deploy_source(
    name="faker-prod",
    source=local_source,
)
cloud_source.check(raise_on_error=True)

# List all connections in the workspace
for connection in workspace.list_connections():
    print(connection.connection_id, connection.name)

# Get and run an existing connection
connection = workspace.get_connection(connection_id="abc-123")
sync_result = connection.run_sync(wait=True, wait_timeout=600)
print(sync_result.status)        # "succeeded" | "failed" | "running"
print(sync_result.records_synced)

# Deploy a full pipeline from source → destination
cloud_dest = workspace.deploy_destination(
    name="snowflake-prod",
    destination=ab.get_destination("destination-snowflake", config={...}),
)
connection = workspace.deploy_connection(
    name="faker-to-snowflake",
    source=cloud_source,
    destination=cloud_dest,
    streams=["users", "purchases"],
)
```

--------------------------------

### Manage Secrets with get_secret

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

Demonstrates using `get_secret` to retrieve sensitive information from various sources like environment variables, .env files, or interactive prompts. It also shows how to register custom secret managers.

```python
import airbyte as ab
from airbyte.secrets import (
    get_secret,
    SecretSourceEnum,
    register_secret_manager,
    GoogleGSMSecretManager,
    EnvVarSecretManager,
)

# Get from any available source (env → .env → Colab → prompt)
token = ab.get_secret("GITHUB_PERSONAL_ACCESS_TOKEN")

# Restrict to environment variables only (no prompt)
token = get_secret(
    "GITHUB_PERSONAL_ACCESS_TOKEN",
    sources=[SecretSourceEnum.ENV],
    allow_prompt=False,
)

# Register Google Secret Manager as a source
gsm = GoogleGSMSecretManager(
    project="my-gcp-project",
    credentials_json=get_secret("GOOGLE_APPLICATION_CREDENTIALS_JSON"),
)
register_secret_manager(gsm)

# Secret references in configs (resolved at runtime)
source = ab.get_source(
    "source-github",
    config={
        "credentials": {
            "personal_access_token": "secret_reference::GITHUB_PERSONAL_ACCESS_TOKEN"
        },
        "repositories": ["airbytehq/pyairbyte"],
    },
)
source.check()  # PyAirbyte auto-resolves the secret_reference:: prefix
```

--------------------------------

### Generate Benchmark Source Records

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

Creates a source that generates synthetic records for performance benchmarking. Requires Docker. Supports custom record counts using scientific notation.

```python
import airbyte as ab
from airbyte.sources.util import get_benchmark_source

# Default: 500,000 records
benchmark_source = get_benchmark_source()

# Custom record count (scientific notation supported)
large_source = get_benchmark_source(num_records="5e6")   # 5 million records

cache = ab.new_local_cache("benchmark_run")
result = large_source.read(cache=cache, force_full_refresh=True)
print(f"Total records: {result.processed_records}")
```

--------------------------------

### Source.get_samples / Source.print_samples

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

Fetches and optionally renders a small number of records from selected streams for quick data preview.

```APIDOC
## `Source.get_samples` / `Source.print_samples` — Quick data preview

### Description
Fetches a small number of records (default 5) from each selected stream. `print_samples` renders them as a rich formatted table in the terminal.

### Method
`get_samples(streams: list[str], limit: int = 5) -> dict[str, InMemoryDataset]`
`print_samples(streams: str | list[str], limit: int = 5, on_error: str = "log")`

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```python
import airbyte as ab

source = ab.get_source("source-faker", config={"count": 100},
                       streams=["users", "products"])

# Get samples as a dict of stream -> InMemoryDataset
samples = source.get_samples(streams=["users", "products"], limit=3)
for stream, dataset in samples.items():
    print(f"{stream}: {list(dataset)}")

# Print rich formatted tables to the terminal
source.print_samples(streams="*", limit=5, on_error="log")
```

### Response
#### Success Response (200)
`get_samples`: A dictionary mapping stream names to `InMemoryDataset` objects containing the sample records.
`print_samples`: None (prints to terminal).

#### Response Example
(Output depends on the data in the streams)
```

--------------------------------

### Programmatic MCP Tool Usage: Sync Source to Cache and Query

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

Synchronizes data from a source connector to a local cache and then executes a SQL query against the cached data using `sync_source_to_cache` and `run_sql_query`.

```python
# Sync and query
sync_source_to_cache(
    source_connector_name="source-faker",
    config={"count": 500},
    streams=["users"],
)
results = run_sql_query("SELECT id, name FROM main.users LIMIT 5")
for row in results:
    print(row)
```

--------------------------------

### Using Caches with Sources and Iterating Streams

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

Shows how to use a cache with an Airbyte source and iterate through streams within a cache. The `read` method accepts a cache object, and caches are iterable.

```python
source = ab.get_source("source-faker", config={\"count\": 5000},
                       streams=[\"users\"])
result = source.read(cache=snowflake_cache)
df = snowflake_cache.get_pandas_dataframe(\"users\")

# Iterate streams in any cache
for stream_name, dataset in pg_cache:
    print(f"{stream_name}: {len(dataset)} rows")
```

--------------------------------

### SQL Cache Backends (SnowflakeCache, PostgresCache, etc.)

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

Provides SQL cache backends that serve as drop-in replacements for `DuckDBCache`, storing data directly in cloud or self-hosted SQL warehouses.

```APIDOC
## `SnowflakeCache` / `PostgresCache` / `BigQueryCache` / `MotherDuckCache` — SQL cache backends

### Description
Drop-in replacements for `DuckDBCache` that land data directly in cloud or self-hosted SQL warehouses. All share the `CacheBase` interface.

### Method
(Specific constructors for each cache type, e.g., `SnowflakeCache(...)`)

### Parameters
(Parameters vary by cache type, typically include connection details like account, username, password, database, schema, etc.)

### Request Example
```python
import airbyte as ab
from airbyte.caches import SnowflakeCache, PostgresCache, BigQueryCache

# Snowflake
snowflake_cache = SnowflakeCache(
    account="myaccount",
    username="myuser",
    password=ab.get_secret("SNOWFLAKE_PASSWORD"),
    warehouse="COMPUTE_WH",
    database="AIRBYTE",
    role="SYSADMIN",
    schema_name="raw",
)

# Usage would be similar to DuckDBCache, e.g.:
# source.read(cache=snowflake_cache)
# df = snowflake_cache.get_pandas_dataframe("users")
```

### Response
#### Success Response (200)
Methods like `get_pandas_dataframe`, `get_arrow_dataset`, and `run_sql_query` return data structures or query results. Direct access via `cache[stream_name]` returns an `InMemoryDataset`.

#### Response Example
(Data structures or query results depending on the method called)
```

--------------------------------

### SQL Cache Backends (Snowflake, Postgres, BigQuery, MotherDuck)

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

Provides drop-in replacements for DuckDBCache that store data directly in cloud or self-hosted SQL warehouses. All share the CacheBase interface.

```python
import airbyte as ab
from airbyte.caches import SnowflakeCache, PostgresCache, BigQueryCache

# Snowflake
snowflake_cache = SnowflakeCache(
    account="myaccount",
    username="myuser",
    password=ab.get_secret("SNOWFLAKE_PASSWORD"),
    warehouse="COMPUTE_WH",
    database="AIRBYTE",
    role="SYSADMIN",
    schema_name="raw",
)
```

--------------------------------

### get_benchmark_source

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

Creates a source that generates synthetic records for benchmarking sink throughput. Requires Docker.

```APIDOC
## `get_benchmark_source` — Performance benchmarking

Creates a source that generates synthetic records for benchmarking sink throughput. Requires Docker.

```python
import airbyte as ab
from airbyte.sources.util import get_benchmark_source

# Default: 500,000 records
benchmark_source = get_benchmark_source()

# Custom record count (scientific notation supported)
large_source = get_benchmark_source(num_records="5e6")   # 5 million records

cache = ab.new_local_cache("benchmark_run")
result = large_source.read(cache=cache, force_full_refresh=True)
print(f"Total records: {result.processed_records}")
```
```

--------------------------------

### Source.read

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

Executes the connector and writes all selected streams into the specified cache. Supports incremental sync or forced full refresh, with configurable write strategies.

```APIDOC
## Source.read — Read data into a cache

Executes the connector and writes all selected streams into the given cache. Returns a `ReadResult` mapping stream names to `CachedDataset` objects. Supports incremental sync (default) or forced full refresh. Write strategy controls upsert/append/replace behavior.

### Parameters

- **cache** (Cache object, optional) - The cache to write the data to. Defaults to a local DuckDB cache.
- **write_strategy** (string, optional) - Controls how data is written to the cache. Options: "append", "merge", "replace", "auto". Defaults to "auto".
- **force_full_refresh** (bool, optional) - If True, ignores existing state and performs a full refresh. Defaults to False.

### Returns

- **ReadResult**: An object containing the results of the read operation, including processed records and a mapping of stream names to `CachedDataset` objects.

### `ReadResult` object attributes

- **processed_records** (int): The total number of records processed.
- **streams** (dict): A dictionary mapping stream names to `CachedDataset` objects.
```

--------------------------------

### Instantiate a Destination Connector

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

Returns a Destination object for any Airbyte destination connector. Supports Docker execution and writing data directly from sources or pre-read results.

```python
import airbyte as ab

# Destination via Docker (typical for Java-based connectors)
destination = ab.get_destination(
    "destination-snowflake",
    config={
        "host": "myaccount.snowflakecomputing.com",
        "role": "MYROLE",
        "warehouse": "MYWAREHOUSE",
        "database": "MYDB",
        "schema": "PUBLIC",
        "username": "myuser",
        "credentials": {
            "password": ab.get_secret("SNOWFLAKE_PASSWORD"),
        },
    },
    docker_image=True,
)

# Write directly from a source (caches intermediately in DuckDB by default)
source = ab.get_source("source-faker", config={"count": 5000}, streams=["users"])
write_result = destination.write(
    source,
    streams=["users"],
    write_strategy="merge",
    force_full_refresh=False,
)
print(f"Records delivered: {write_result.processed_records}")

# Write from a pre-read result (skip caching)
read_result = source.read()
write_result = destination.write(read_result, cache=False)
```

--------------------------------

### get_available_connectors

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

Returns a list of available connector names from the Airbyte connector registry, optionally filtered by type.

```APIDOC
## `get_available_connectors` — Registry lookup

Returns a list of available connector names from the Airbyte connector registry, optionally filtered by type.

```python
import airbyte as ab

# All connectors
all_connectors = ab.get_available_connectors()
print(f"Total connectors: {len(all_connectors)}")

# Filter by source or destination
sources = ab.get_available_connectors(connector_type="source")
destinations = ab.get_available_connectors(connector_type="destination")

```

--------------------------------

### Programmatic MCP Tool Usage: List Source Streams

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

Lists the available streams for a given source connector and configuration using the `list_source_streams` function.

```python
# List streams
streams = list_source_streams(
    source_connector_name="source-faker",
    config={"count": 100},
)
print(streams)  # ['products', 'purchases', 'users']
```

--------------------------------

### Create Local and Colab Caches

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

Provides factory functions to create default, named local, or Google Colab persistent caches. Local caches can be configured with cleanup options.

```python
import airbyte as ab

# Default cache: .cache/default_cache/default_cache.duckdb
default = ab.get_default_cache()

# Named local cache with a unique identifier
my_cache = ab.new_local_cache(
    cache_name="etl_pipeline",   # alphanumeric + underscores only
    cache_dir=".cache/etl",
    cleanup=True,
)

# Google Colab persistent cache (mounts Google Drive)
colab_cache = ab.get_colab_cache(
    cache_name="project_cache",
    sub_dir="Airbyte/my_project",
    drive_name="MyDrive",      # or shared drive name
)

source = ab.get_source("source-faker", config={\"count\": 1000})
source.select_all_streams()
result = source.read(cache=my_cache)
```

--------------------------------

### Manage CloudConnection lifecycle

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

Control a single Airbyte Cloud connection, including running syncs, managing state, scheduling, and performing CRUD operations. Ensure the workspace is initialized first.

```python
from airbyte import cloud
import airbyte as ab

workspace = cloud.CloudWorkspace(
    workspace_id=ab.get_secret("AIRBYTE_CLOUD_WORKSPACE_ID"),
    client_id=ab.get_secret("AIRBYTE_CLOUD_CLIENT_ID"),
    client_secret=ab.get_secret("AIRBYTE_CLOUD_CLIENT_SECRET"),
)
connection = workspace.get_connection("my-connection-id")

# Run a sync and wait for it to complete
sync_result = connection.run_sync(wait=True, wait_timeout=300)

# Retrieve recent job history
logs = connection.get_previous_sync_logs(limit=5)
for log in logs:
    print(log.job_id, log.status, log.records_synced)

# Enable/disable and reschedule
connection.enabled = True
connection.set_schedule("0 0 * * *")    # daily at midnight UTC
connection.set_manual_schedule()        # disable automatic scheduling

# Inspect and modify stream selection
print(connection.stream_names)
connection.set_selected_streams(["users", "purchases"])

# State management
state = connection.dump_raw_state()         # Airbyte protocol format
stream_state = connection.get_stream_state("users")
connection.set_stream_state("users", {"cursor": "2024-06-01T00:00:00Z"})
connection.import_raw_state(state)          # restore a previous state

# Rename / set table prefix
connection.rename("my-renamed-connection")
connection.set_table_prefix("raw_")

# Cleanup
connection.permanently_delete(
    cascade_delete_source=False,
    cascade_delete_destination=False,
)
```

--------------------------------

### Find GitHub Connectors

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

Filters a list of sources to find those related to GitHub.

```python
github_connectors = [c for c in sources if "github" in c.lower()]
print(github_connectors)
```

--------------------------------

### Serve MCP Locally

Source: https://github.com/airbytehq/pyairbyte/blob/main/docs/CONTRIBUTING.md

Helper Poe tasks to serve the MCP server locally using different transport methods: STDIO, HTTP, or Server-Sent Events.

```bash
poe mcp-serve-local    # STDIO transport (default)
poe mcp-serve-http     # HTTP transport on localhost:8000
poe mcp-serve-sse      # Server-Sent Events transport on localhost:8000

poe mcp-inspect        # Show all available MCP tools and their schemas
```

--------------------------------

### Access and Convert Cached Datasets

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

Demonstrates accessing `CachedDataset` objects, iterating them, and converting them to pandas DataFrames, PyArrow Datasets, or SQLAlchemy Tables. Supports SQL filtering and document conversion.

```python
import airbyte as ab

source = ab.get_source("source-faker", config={\"count\": 5000},
                       streams=[\"users\", \"purchases\"])
result = source.read()

# Get a dataset
users: ab.CachedDataset = result[\"users\"]

# Iteration (lazy, row by row)
for record in users:
    print(record[\"id\"], record[\"name\"])

# Convert to pandas
df = users.to_pandas()

# Convert to PyArrow Dataset (chunked)
arrow_ds = users.to_arrow(max_chunk_size=5_000)

# Get underlying SQLAlchemy Table object
sql_table = users.to_sql_table()

# Apply SQL filters (returns new SQLDataset, lazy)
from sqlalchemy import column
filtered = users.with_filter("age > 25", column(\"name\").like(\"A%\"))
print(len(filtered))

# Column introspection
print(users.column_names)

# Documents for AI/LLM pipelines
from airbyte.documents import Document
docs = list(users.to_documents(
    title_property=\"name\",
    content_properties=[\"email\", \"name\"],
    metadata_properties=[\"id\"]
))
print(docs[0].page_content)
```

--------------------------------

### Select Streams for Data Reading

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

Use `select_streams` or `select_all_streams` to specify which streams to read. Raises an error if a stream is not found in the catalog. Allows overriding cursor and primary keys per stream.

```python
import airbyte as ab

source = ab.get_source("source-faker", config={"count": 500})

# Select specific streams
source.select_streams(["users", "purchases"])

# Or select everything
source.select_all_streams()

# Check what is selected
print(source.get_selected_streams())
# ['users', 'purchases']

# Override cursor / primary key per stream
source.set_cursor_keys(users="updated_at")
source.set_primary_keys(users="id", purchases=["user_id", "purchase_id"])
```

--------------------------------

### Pin GitHub Actions SHA

Source: https://github.com/airbytehq/pyairbyte/blob/main/docs/CONTRIBUTING.md

Use the `pinact` tool to convert fixed GitHub Actions versions to their SHA-pinned equivalents for enhanced security. Specify an optional file to process.

```bash
# Convert from from fixed version to sha
# Example: actions/checkout@v4 -> actions/checkout@08e... # v4.3.0
pinact run [optional_file]
```

--------------------------------

### Convert Source Records to LLM Documents

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

Converts records from a source stream into Document objects compatible with LangChain and similar AI frameworks. Specify fields for title, content, and metadata.

```python
import airbyte as ab

source = ab.get_source(
    "source-github",
    config={"credentials": {"personal_access_token": ab.get_secret("GITHUB_PAT")}},
            "repositories": ["airbytehq/pyairbyte"]},
    streams=["issues"],
)

docs = source.get_documents(
    stream="issues",
    title_property="title",          # field to use as document title
    content_properties=["body"],     # fields that form the main text
    metadata_properties=["number", "state", "html_url"],
    render_metadata=False,
)

for doc in docs:
    print(doc.id)
    print(doc.page_content[:200])    # LangChain-compatible alias for content
    print(doc.metadata)
```

--------------------------------

### Process SyncResult from Cloud sync

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

Examine the outcome of a remote Airbyte Cloud sync job. Access status, metrics, logs, and directly query data for supported destinations.

```python
from airbyte import cloud
import airbyte as ab

workspace = cloud.CloudWorkspace(
    workspace_id=ab.get_secret("AIRBYTE_CLOUD_WORKSPACE_ID"),
    client_id=ab.get_secret("AIRBYTE_CLOUD_CLIENT_ID"),
    client_secret=ab.get_secret("AIRBYTE_CLOUD_CLIENT_SECRET"),
)
connection = workspace.get_connection("my-connection-id")
sync_result = connection.run_sync(wait=True)

# Status and metrics
print(sync_result.status)            # "succeeded"
print(sync_result.records_synced)
print(sync_result.bytes_synced)
print(sync_result.start_time)

# Wait for completion manually (if run with wait=False)
async_sync = connection.run_sync(wait=False)
async_sync.wait_for_completion(wait_timeout=600, raise_failure=True)

# Retrieve full log text
print(async_sync.get_full_log_text())

# For supported destinations: read data from the sync result
dataset = sync_result.get_dataset("users")
for record in dataset:
    print(record)
df = dataset.to_pandas()
```

--------------------------------

### get_destination

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

Instantiates a `Destination` object for any Airbyte destination connector, supporting various execution options.

```APIDOC
## `get_destination` — Instantiate a destination connector

### Description
Returns a `Destination` object for any Airbyte destination connector. Most destinations require Docker. Supports all the same execution options as `get_source`.

### Method
`get_destination(connector_name: str, config: dict, docker_image: bool = False)`

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```python
import airbyte as ab

# Destination via Docker (typical for Java-based connectors)
destination = ab.get_destination(
    "destination-snowflake",
    config={
        "host": "myaccount.snowflakecomputing.com",
        "role": "MYROLE",
        "warehouse": "MYWAREHOUSE",
        "database": "MYDB",
        "schema": "PUBLIC",
        "username": "myuser",
        "credentials": {
            "password": ab.get_secret("SNOWFLAKE_PASSWORD"),
        },
    },
    docker_image=True,
)

# Write directly from a source (caches intermediately in DuckDB by default)
source = ab.get_source("source-faker", config={"count": 5000}, streams=["users"])
write_result = destination.write(
    source,
    streams=["users"],
    write_strategy="merge",
    force_full_refresh=False,
)
print(f"Records delivered: {write_result.processed_records}")

# Write from a pre-read result (skip caching)
read_result = source.read()
write_result = destination.write(read_result, cache=False)
```

### Response
#### Success Response (200)
A `Destination` object configured for the specified connector.

#### Response Example
(Destination object)
```

--------------------------------

### CloudConnection

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

Manages a single source→destination EL connection in Airbyte Cloud. Supports running syncs, managing state, scheduling, and CRUD operations.

```APIDOC
## `CloudConnection` — Cloud connection lifecycle management

Manages a single source→destination EL connection in Airbyte Cloud. Supports running syncs, managing state, scheduling, and CRUD operations.

```python
from airbyte import cloud
import airbyte as ab

workspace = cloud.CloudWorkspace(
    workspace_id=ab.get_secret("AIRBYTE_CLOUD_WORKSPACE_ID"),
    client_id=ab.get_secret("AIRBYTE_CLOUD_CLIENT_ID"),
    client_secret=ab.get_secret("AIRBYTE_CLOUD_CLIENT_SECRET"),
)
connection = workspace.get_connection("my-connection-id")

# Run a sync and wait for it to complete
sync_result = connection.run_sync(wait=True, wait_timeout=300)

# Retrieve recent job history
logs = connection.get_previous_sync_logs(limit=5)
for log in logs:
    print(log.job_id, log.status, log.records_synced)

# Enable/disable and reschedule
connection.enabled = True
connection.set_schedule("0 0 * * *")    # daily at midnight UTC
connection.set_manual_schedule()        # disable automatic scheduling

# Inspect and modify stream selection
print(connection.stream_names)
connection.set_selected_streams(["users", "purchases"])

# State management
state = connection.dump_raw_state()         # Airbyte protocol format
stream_state = connection.get_stream_state("users")
connection.set_stream_state("users", {"cursor": "2024-06-01T00:00:00Z"})
connection.import_raw_state(state)          # restore a previous state

# Rename / set table prefix
connection.rename("my-renamed-connection")
connection.set_table_prefix("raw_")

# Cleanup
connection.permanently_delete(
    cascade_delete_source=False,
    cascade_delete_destination=False,
)
```
```

--------------------------------

### ReadResult Object Usage

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

Explains how to use the `ReadResult` object returned by `Source.read()`. It provides access to stream datasets, total processed records, and cache/SQL engine information.

```python
import airbyte as ab

source = ab.get_source("source-faker", config={\"count\": 2000},
                       streams=[\"users\", \"products\", \"purchases\"])
result = source.read()

# Total records across all streams
print(f"Total records: {result.processed_records}")

# Stream names that were processed
print(list(result))           # [\'users\', \'products\', \'purchases\']
print(\"users\" in result)      # True

# Access stream dataset
purchases_df = result[\"purchases\"].to_pandas()

# The underlying cache
cache = result.cache
engine = result.get_sql_engine()   # SQLAlchemy Engine

# Streams mapping
for name, dataset in result.streams.items():
    print(f"{name}: {len(dataset)} records")
```

--------------------------------

### Source.select_streams / Source.select_all_streams

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

Selects which streams to include in the next `read()` call. Accepts a list of stream names or `"*"` for all. Raises `AirbyteStreamNotFoundError` if a stream name is not in the discovered catalog.

```APIDOC
## Source.select_streams / Source.select_all_streams — Stream selection

Selects which streams to include in the next `read()` call. Accepts a list of stream names or `"*"` for all. Raises `AirbyteStreamNotFoundError` if a stream name is not in the discovered catalog.

### Parameters

- **stream_names** (list[string] or "*") - Required - A list of stream names to select, or `"*"` to select all available streams.

### Methods on Source object

- **get_selected_streams()**: Returns a list of currently selected streams.
- **set_cursor_keys(**stream_name**=key, ...)**: Sets the cursor field for a specific stream.
- **set_primary_keys(**stream_name**=keys, ...)**: Sets the primary key fields for a specific stream.
```

--------------------------------

### DuckDBCache

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

Provides a local DuckDB cache for storing and querying data, used automatically by `source.read()` when no cache is specified.

```APIDOC
## `DuckDBCache` — Local DuckDB cache

### Description
The default cache backend, storing data in a local DuckDB file. Used automatically by `source.read()` when no cache is specified. Supports SQL queries, pandas/Arrow export, and stream iteration.

### Method
`DuckDBCache(db_path: str, schema_name: str = "public", table_prefix: str = "")`

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```python
import airbyte as ab
from airbyte.caches import DuckDBCache

# Named persistent cache
cache = DuckDBCache(
    db_path=".cache/myproject/myproject.duckdb",
    schema_name="raw_data",
    table_prefix="src_",
)

# Use it in a read
source = ab.get_source("source-faker", config={"count": 1000},
                       streams=["users", "purchases"])
result = source.read(cache=cache)

# Access streams from the cache directly
users_dataset = cache["users"]
print(len(users_dataset))            # row count

# Export to pandas / Arrow
df = cache.get_pandas_dataframe("users")
arrow_ds = cache.get_arrow_dataset("users", max_chunk_size=10_000)

# Run arbitrary SQL
rows = cache.run_sql_query(
    "SELECT id, name FROM raw_data.src_users WHERE age > 30 LIMIT 10"
)
for row in rows:
    print(row)

# Context manager ensures connections are closed
with DuckDBCache(db_path=".cache/temp.duckdb") as temp_cache:
    source.read(cache=temp_cache)
    df = temp_cache.get_pandas_dataframe("users")
```

### Response
#### Success Response (200)
Methods like `get_pandas_dataframe`, `get_arrow_dataset`, and `run_sql_query` return data structures or query results. Direct access via `cache[stream_name]` returns an `InMemoryDataset`.

#### Response Example
(Data structures or query results depending on the method called)
```

--------------------------------

### Read Data into a Cache

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

The `read` method executes the connector and writes selected streams to a cache. Supports incremental sync or full refresh, and various write strategies. Results can be accessed as `CachedDataset` objects or converted to pandas DataFrames.

```python
import airbyte as ab

source = ab.get_source(
    "source-faker",
    config={"count": 10_000},
    streams=["users", "purchases"],
)
source.check()

# Default: uses local DuckDB cache at .cache/default_cache/
result = source.read()

# With a named local DuckDB cache and incremental merge
custom_cache = ab.new_local_cache("my_project")
result = source.read(
    cache=custom_cache,
    write_strategy="merge",      # "append" | "merge" | "replace" | "auto"
    force_full_refresh=False,    # True = ignore existing state
)

# Inspect results
print(f"Records read: {result.processed_records}")
for stream_name, dataset in result.streams.items():
    print(f"  {stream_name}: {len(dataset)} records")

# Access a specific stream
users_df = result["users"].to_pandas()
print(users_df.head())
```

--------------------------------

### Local DuckDB Cache for Data Storage

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

Utilizes a local DuckDB file for caching data, supporting SQL queries, pandas/Arrow export, and stream iteration. Can be used as a context manager for automatic connection closing.

```python
import airbyte as ab
from airbyte.caches import DuckDBCache

# Named persistent cache
cache = DuckDBCache(
    db_path=".cache/myproject/myproject.duckdb",
    schema_name="raw_data",
    table_prefix="src_",
)

# Use it in a read
source = ab.get_source("source-faker", config={"count": 1000},
                       streams=["users", "purchases"])
result = source.read(cache=cache)

# Access streams from the cache directly
users_dataset = cache["users"]
print(len(users_dataset))            # row count

# Export to pandas / Arrow
df = cache.get_pandas_dataframe("users")
arrow_ds = cache.get_arrow_dataset("users", max_chunk_size=10_000)

# Run arbitrary SQL
rows = cache.run_sql_query(
    "SELECT id, name FROM raw_data.src_users WHERE age > 30 LIMIT 10"
)
for row in rows:
    print(row)

# Context manager ensures connections are closed
with DuckDBCache(db_path=".cache/temp.duckdb") as temp_cache:
    source.read(cache=temp_cache)
    df = temp_cache.get_pandas_dataframe("users")
```

--------------------------------

### Programmatic MCP Tool Usage: Validate Connector Config

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

Validates a connector configuration using the `validate_connector_config` function from the `airbyte.mcp.local` module. Returns a boolean indicating validity and a message.

```python
# Programmatic use of MCP tools (for testing)
from airbyte.mcp.local import (
    validate_connector_config,
    list_source_streams,
    read_source_stream_records,
    sync_source_to_cache,
    run_sql_query,
)

# Validate a config
ok, msg = validate_connector_config(
    connector_name="source-faker",
    config={"count": 100},
)
print(ok, msg)  # True, "Configuration for source-faker is valid!"
```

--------------------------------

### Source.get_documents

Source: https://context7.com/airbytehq/pyairbyte/llms.txt

Converts records from a source stream into `Document` objects suitable for AI frameworks like LangChain. Documents include `id`, `content`, `metadata`, and `page_content`.

```APIDOC
## `Source.get_documents` — Records as LLM-ready Documents

### Description
Converts records from a source stream into `Document` objects compatible with LangChain and similar AI frameworks. Each document has `id`, `content`, `metadata`, and `page_content` fields.

### Method
`get_documents(
    stream: str,
    title_property: str = None,
    content_properties: list[str] = None,
    metadata_properties: list[str] = None,
    render_metadata: bool = True
)`

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```python
import airbyte as ab

source = ab.get_source(
    "source-github",
    config={"credentials": {"personal_access_token": ab.get_secret("GITHUB_PAT")}},
            "repositories": ["airbytehq/pyairbyte"],
    streams=["issues"],
)

docs = source.get_documents(
    stream="issues",
    title_property="title",          # field to use as document title
    content_properties=["body"],     # fields that form the main text
    metadata_properties=["number", "state", "html_url"],
    render_metadata=False,
)

for doc in docs:
    print(doc.id)
    print(doc.page_content[:200])    # LangChain-compatible alias for content
    print(doc.metadata)
```

### Response
#### Success Response (200)
An iterable of `Document` objects.

#### Response Example
(Iterable of Document objects, each with id, content, metadata, page_content)
```