ADBC¶

Arrow Database Connectivity (ADBC) provides a vendor-neutral API for database access using Apache Arrow’s columnar format. ADBC enables zero-copy data transfer and high-performance analytics across multiple database backends.

Overview¶

The ADBC ADK store supports multiple database backends through ADBC drivers:

PostgreSQL - Production-ready with full feature support
SQLite - Development and testing
DuckDB - Embedded analytics
Flight SQL - Distributed query execution
Snowflake - Cloud data warehouse

Key Benefits:

Zero-Copy Data Transfer - Arrow-native data exchange eliminates serialization overhead
Columnar Format - Efficient for analytical workloads
Vendor Neutral - Single API across multiple databases
High Performance - Optimized for large-scale data operations

Use Cases:

Analytical AI agents processing large datasets
Multi-backend applications requiring database portability
Integration with Arrow ecosystem (Polars, PyArrow, etc.)
Bulk data operations and ETL pipelines

Installation¶

Install ADBC with your chosen driver:

PostgreSQL:

pip install sqlspec[adbc-postgresql]
# or
uv add 'sqlspec[adbc-postgresql]'

SQLite:

pip install sqlspec[adbc-sqlite]
# or
uv add 'sqlspec[adbc-sqlite]'

DuckDB:

pip install sqlspec[adbc-duckdb]
# or
uv add 'sqlspec[adbc-duckdb]'

Quick Start¶

Basic SQLite Example¶

from sqlspec.adapters.adbc import AdbcConfig
from sqlspec.adapters.adbc.adk import AdbcADKStore

# Configure ADBC with SQLite backend
config = AdbcConfig(connection_config={
    "driver_name": "sqlite",
    "uri": "file:agent.db"
})

# Initialize store and create tables
store = AdbcADKStore(config)
store.create_tables()

# Create session
session = store.create_session(
    session_id="session-1",
    app_name="my-agent",
    user_id="user-123",
    state={"step": 1, "context": "initialized"}
)

# Create event
event = store.create_event(
    event_id="event-1",
    session_id="session-1",
    app_name="my-agent",
    user_id="user-123",
    author="assistant",
    content={"message": "Processing request..."}
)

PostgreSQL Production Example¶

from sqlspec.adapters.adbc import AdbcConfig
from sqlspec.adapters.adbc.adk import AdbcADKStore

# Configure ADBC with PostgreSQL backend
config = AdbcConfig(connection_config={
    "driver_name": "postgresql",
    "uri": "postgresql://user:pass@localhost:5432/agentdb",
    "username": "agent_user",
    "password": "secure_password"
})

store = AdbcADKStore(config)
store.create_tables()

# Sessions and events work identically across backends
session = store.create_session(
    session_id="prod-session-1",
    app_name="production-agent",
    user_id="user-456",
    state={"environment": "production", "version": "1.0"}
)

Configuration¶

ADBC Connection Parameters¶

from sqlspec.adapters.adbc import AdbcConfig

config = AdbcConfig(
    connection_config={
        "driver_name": "postgresql",  # or "sqlite", "duckdb", etc.
        "uri": "postgresql://host:port/database",
        "username": "user",
        "password": "pass",
        # Driver-specific options
        "adbc.connection.autocommit": "true"
    }
)

Custom Table Names¶

store = AdbcADKStore(
    config,
    session_table="agent_sessions",
    events_table="agent_events"
)

Schema¶

Sessions Table¶

CREATE TABLE IF NOT EXISTS adk_sessions (
    id VARCHAR(128) PRIMARY KEY,
    app_name VARCHAR(128) NOT NULL,
    user_id VARCHAR(128) NOT NULL,
    state TEXT NOT NULL DEFAULT '{}',
    create_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
    update_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
)

CREATE INDEX IF NOT EXISTS idx_adk_sessions_app_user
    ON adk_sessions(app_name, user_id);

CREATE INDEX IF NOT EXISTS idx_adk_sessions_update_time
    ON adk_sessions(update_time DESC);

Events Table¶

CREATE TABLE IF NOT EXISTS adk_events (
    id VARCHAR(128) PRIMARY KEY,
    session_id VARCHAR(128) NOT NULL,
    app_name VARCHAR(128) NOT NULL,
    user_id VARCHAR(128) NOT NULL,
    invocation_id VARCHAR(256),
    author VARCHAR(256),
    actions BLOB,
    long_running_tool_ids_json TEXT,
    branch VARCHAR(256),
    timestamp TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
    content TEXT,
    grounding_metadata TEXT,
    custom_metadata TEXT,
    partial INTEGER,
    turn_complete INTEGER,
    interrupted INTEGER,
    error_code VARCHAR(256),
    error_message VARCHAR(1024),
    FOREIGN KEY (session_id) REFERENCES adk_sessions(id) ON DELETE CASCADE
)

CREATE INDEX IF NOT EXISTS idx_adk_events_session
    ON adk_events(session_id, timestamp ASC);

Field Types:

TEXT - JSON-serialized data (state, content, metadata)
BLOB - Pickled actions from Google ADK
INTEGER - Boolean fields (0/1/NULL)
TIMESTAMP - Created/updated timestamps
VARCHAR - String identifiers with length limits

Usage Patterns¶

Session Management¶

# Create session
session = store.create_session(
    session_id="unique-id",
    app_name="my-agent",
    user_id="user-123",
    state={"conversation": [], "context": {}}
)

# Get session
session = store.get_session("unique-id")
if session:
    print(session["state"])

# Update session state
store.update_session_state("unique-id", {
    "conversation": [...],
    "context": {...},
    "updated": True
})

# List user's sessions
sessions = store.list_sessions("my-agent", "user-123")
for session in sessions:
    print(f"{session['id']}: {session['state']}")

# Delete session (cascades to events)
store.delete_session("unique-id")

Event Management¶

# Create event with all fields
event = store.create_event(
    event_id="event-123",
    session_id="session-id",
    app_name="my-agent",
    user_id="user-123",
    author="assistant",
    actions=b"pickled_actions",
    content={"message": "Response text"},
    grounding_metadata={"sources": ["doc1", "doc2"]},
    custom_metadata={"confidence": 0.95},
    partial=False,
    turn_complete=True,
    interrupted=False
)

# List session events (ordered by timestamp)
events = store.list_events("session-id")
for event in events:
    print(f"{event['timestamp']}: {event['content']}")

Database-Specific Notes¶

SQLite¶

Advantages:

Simple setup for development
Serverless (embedded database)
Good for single-user agents

Limitations:

Foreign key enforcement requires PRAGMA foreign_keys = ON per connection
ADBC creates new connections per operation, so cascade deletes may not work reliably
No concurrent writes (single writer)

Recommendation: Use SQLite for development/testing, PostgreSQL for production.

PostgreSQL¶

Advantages:

Full ACID compliance
Excellent concurrency
JSON/JSONB support for efficient queries
Production-grade reliability

Recommendation: Best choice for production AI agents.

DuckDB¶

Advantages:

Embedded analytical database
Excellent for processing large datasets
Arrow-native with zero-copy integration
SQL analytics capabilities

Use Cases: Agents performing data analysis, reporting, or aggregations.

Performance Considerations¶

Connection Management¶

ADBC creates a new connection for each operation by default. For high-throughput applications:

Use connection pooling at the application level
Consider batch operations where possible
Monitor connection creation overhead

Data Types¶

JSON serialization: Uses to_json/from_json for cross-database compatibility
Arrow format: Data returned as Arrow Tables/RecordBatches for zero-copy access
BLOB storage: Actions are stored as binary data (pickled by Google ADK)

Indexing¶

The ADK store creates indexes on:

(app_name, user_id) for session listing
update_time DESC for recent session queries
(session_id, timestamp ASC) for event chronology

Migration from Other Adapters¶

ADBC uses standard SQL compatible with most databases. To migrate:

Export data from existing store
Configure ADBC with your target database
Create tables using store.create_tables()
Import data using standard SQL or bulk insert operations

# Example: Migrate from AsyncPG to ADBC
from sqlspec.adapters.asyncpg import AsyncpgConfig
from sqlspec.adapters.asyncpg.adk import AsyncpgADKStore
from sqlspec.adapters.adbc import AdbcConfig
from sqlspec.adapters.adbc.adk import AdbcADKStore

# Source (AsyncPG)
source_config = AsyncpgConfig(pool_config={"dsn": "..."})
source_store = AsyncpgADKStore(source_config)

# Destination (ADBC)
dest_config = AdbcConfig(connection_config={
    "driver_name": "postgresql",
    "uri": "..."
})
dest_store = AdbcADKStore(dest_config)
dest_store.create_tables()

# Migrate sessions
async for session in source_store.list_sessions("app", "user"):
    dest_store.create_session(
        session_id=session["id"],
        app_name=session["app_name"],
        user_id=session["user_id"],
        state=session["state"]
    )

Troubleshooting¶

Foreign Key Constraints¶

If cascade deletes don’t work with SQLite:

# Manually enable foreign keys for SQLite
with config.provide_connection() as conn:
    cursor = conn.cursor()
    try:
        cursor.execute("PRAGMA foreign_keys = ON")
        conn.commit()
    finally:
        cursor.close()

Driver Not Found¶

Ensure you’ve installed the correct ADBC driver:

# PostgreSQL
pip install adbc-driver-postgresql

# SQLite
pip install adbc-driver-sqlite

# DuckDB
pip install adbc-driver-duckdb

Connection Errors¶

Verify connection string format for your driver:

SQLite: "sqlite:///path/to/db.sqlite" or "file:/path/to/db.sqlite"
PostgreSQL: "postgresql://user:pass@host:port/database"
DuckDB: "duckdb:///path/to/db.duckdb"

API Reference¶

class sqlspec.adapters.adbc.adk.AdbcADKStore[source]

Bases: BaseSyncADKStore[AdbcConfig]

ADBC synchronous ADK store for Arrow Database Connectivity.

Implements session and event storage for Google Agent Development Kit using ADBC. ADBC provides a vendor-neutral API with Arrow-native data transfer across multiple databases (PostgreSQL, SQLite, DuckDB, etc.).

Provides: - Session state management with JSON serialization (TEXT storage) - Event history tracking with BLOB-serialized actions - Timezone-aware timestamps - Foreign key constraints with cascade delete - Database-agnostic SQL (supports multiple backends)

Parameters:: config¶ (AdbcConfig) – AdbcConfig with extension_config[“adk”] settings.

Example

from sqlspec.adapters.adbc import AdbcConfig from sqlspec.adapters.adbc.adk import AdbcADKStore

config = AdbcConfig(

connection_config={“driver_name”: “sqlite”, “uri”: “:memory:”}, extension_config={

“adk”: {
“session_table”: “my_sessions”, “events_table”: “my_events”, “owner_id_column”: “tenant_id INTEGER REFERENCES tenants(id)”

}

}

) store = AdbcADKStore(config) store.create_tables()

Notes

TEXT for JSON storage (compatible across all ADBC backends)
BLOB for pre-serialized actions from Google ADK
TIMESTAMP for timezone-aware timestamps (driver-dependent precision)
INTEGER for booleans (0/1/NULL)
Parameter style varies by backend (?, $1, :name, etc.)
Uses dialect-agnostic SQL for maximum compatibility
State and JSON fields use to_json/from_json for serialization
ADBC drivers handle parameter binding automatically
Configuration is read from config.extension_config[“adk”]

__init__(config)[source]

Initialize ADBC ADK store.

Parameters:: config¶ (AdbcConfig) – AdbcConfig instance (any ADBC driver).

Notes

Configuration is read from config.extension_config[“adk”]: - session_table: Sessions table name (default: “adk_sessions”) - events_table: Events table name (default: “adk_events”) - owner_id_column: Optional owner FK column DDL (default: None)

property dialect: str: Return the detected database dialect.

create_tables()[source]

Create both sessions and events tables if they don’t exist.

Return type:: None

create_session(session_id, app_name, user_id, state, owner_id=None)[source]

Create a new session.

Parameters:

session_id¶ (str) – Unique session identifier.
app_name¶ (str) – Application name.
user_id¶ (str) – User identifier.
state¶ (dict[str, typing.Any]) – Initial session state.
owner_id¶ (Optional[typing.Any]) – Optional owner ID value for owner_id_column (can be None for nullable columns).

Return type:

SessionRecord

Returns:

Created session record.

get_session(session_id)[source]

Get session by ID.

Parameters:: session_id¶ (str) – Session identifier.
Return type:: SessionRecord | None
Returns:: Session record or None if not found.

Notes

State is deserialized from JSON string.

update_session_state(session_id, state)[source]

Update session state.

Parameters:

session_id¶ (str) – Session identifier.
state¶ (dict[str, typing.Any]) – New state dictionary (replaces existing state).

Return type:

None

Notes

This replaces the entire state dictionary. Updates update_time to current timestamp.

delete_session(session_id)[source]

Delete session and all associated events (cascade).

Parameters:: session_id¶ (str) – Session identifier.
Return type:: None

Notes

Foreign key constraint ensures events are cascade-deleted.

list_sessions(app_name, user_id=None)[source]

List sessions for an app, optionally filtered by user.

Parameters:

app_name¶ (str) – Application name.
user_id¶ (str | None) – User identifier. If None, lists all sessions for the app.

Return type:

list[SessionRecord]

Returns:

List of session records ordered by update_time DESC.

Notes

Uses composite index on (app_name, user_id) when user_id is provided.

create_event(event_id, session_id, app_name, user_id, author=None, actions=None, content=None, **kwargs)[source]

Create a new event.

Parameters:

event_id¶ (str) – Unique event identifier.
session_id¶ (str) – Session identifier.
app_name¶ (str) – Application name.
user_id¶ (str) – User identifier.
author¶ (str | None) – Event author (user/assistant/system).
actions¶ (bytes | None) – Pickled actions object.
content¶ (dict[str, typing.Any] | None) – Event content (JSON).
**kwargs¶ (Any) – Additional optional fields.

Return type:

EventRecord

Returns:

Created event record.

Notes

Uses CURRENT_TIMESTAMP for timestamp if not provided. JSON fields are serialized to JSON strings. Boolean fields are converted to INTEGER (0/1).

list_events(session_id)[source]

List events for a session ordered by timestamp.

Parameters:: session_id¶ (str) – Session identifier.
Return type:: list[EventRecord]
Returns:: List of event records ordered by timestamp ASC.

Notes

Uses index on (session_id, timestamp ASC). JSON fields deserialized from JSON strings. Converts INTEGER booleans to Python bool.

property config: ConfigT: Return the database configuration.

property events_table: str: Return the events table name.

property owner_id_column_ddl: str | None: Return the full owner ID column DDL (or None if not configured).

property owner_id_column_name: str | None: Return the owner ID column name only (or None if not configured).

property session_table: str: Return the sessions table name.

ADBC¶

Overview¶

Installation¶

Quick Start¶

Basic SQLite Example¶

PostgreSQL Production Example¶

Configuration¶

ADBC Connection Parameters¶

Custom Table Names¶

Schema¶

Sessions Table¶

Events Table¶

Usage Patterns¶

Session Management¶

Event Management¶

Database-Specific Notes¶

SQLite¶

PostgreSQL¶

DuckDB¶

Performance Considerations¶

Connection Management¶

Data Types¶

Indexing¶

Migration from Other Adapters¶

Troubleshooting¶

Foreign Key Constraints¶

Driver Not Found¶

Connection Errors¶

API Reference¶

See Also¶