Google ADK Integration¶

"""Configure Google ADK session, event, and memory table names on an ADBC adapter."""

from sqlspec.adapters.adbc import AdbcConfig

__all__ = ("main",)


def main() -> AdbcConfig:
    """Attach ADK store table settings via extension_config on AdbcConfig."""
    adk_config = {
        "session_table": "adk_session",
        "events_table": "adk_event",
        "memory_table": "adk_memory",
        "memory_use_fts": True,
    }
    config = AdbcConfig(
        connection_config={"driver_name": "gizmosql", "gizmosql_backend": "duckdb"},
        extension_config={"adk": adk_config},
    )
    adk_settings = config.extension_config.get("adk")
    assert isinstance(adk_settings, dict)
    assert adk_settings.get("memory_use_fts") is True
    return config


if __name__ == "__main__":
    print(main())

"""Persist and reload Google ADK session events using AiosqliteADKStore."""

import asyncio
from datetime import UTC, datetime

from google.adk.events.event import Event
from google.genai import types

from sqlspec.adapters.aiosqlite import AiosqliteConfig
from sqlspec.adapters.aiosqlite.adk import AiosqliteADKStore
from sqlspec.extensions.adk import StoredEvent
from sqlspec.extensions.adk.converters import event_to_record

__all__ = ("main",)


async def main() -> list[StoredEvent]:
    """Create an ADK session, append a user event, and fetch stored events."""
    config = AiosqliteConfig(connection_config={"database": ":memory:"})
    try:
        store = AiosqliteADKStore(config)
        await store.ensure_tables()

        session = await store.create_session(
            session_id="session_1", app_name="docs", user_id="user_1", state={"mode": "demo"}
        )

        event = Event(
            id="evt_1",
            invocation_id="inv_1",
            author="user",
            content=types.Content(parts=[types.Part(text="Hello")]),
            timestamp=datetime.now(UTC).timestamp(),
        )
        event_record = event_to_record(event, session["app_name"], session["user_id"], session["id"])
        await store.append_event(event_record)
        events = await store.get_events(session["app_name"], session["user_id"], session["id"])
        assert len(events) == 1
        return events
    finally:
        await config.close_pool()


if __name__ == "__main__":
    print(len(asyncio.run(main())))

"""Index and query tool-generated memories with full-text search in AiosqliteADKMemoryStore."""

import asyncio
from datetime import UTC, datetime

from sqlspec.adapters.aiosqlite import AiosqliteConfig
from sqlspec.adapters.aiosqlite.adk import AiosqliteADKMemoryStore
from sqlspec.extensions.adk import StoredMemory

__all__ = ("main",)


async def main() -> int:
    """Insert a tool invocation memory entry and search it via FTS."""
    config = AiosqliteConfig(
        connection_config={"database": ":memory:"}, extension_config={"adk": {"memory_use_fts": True}}
    )
    try:
        store = AiosqliteADKMemoryStore(config)
        await store.ensure_tables()

        record: StoredMemory = {
            "id": "mem_1",
            "session_id": "session_1",
            "app_name": "docs",
            "user_id": "user_1",
            "scope": "user",
            "event_id": "evt_1",
            "author": "tool",
            "timestamp": datetime.now(UTC),
            "content_json": {"tool": "search", "query": "sqlspec"},
            "content_text": "tool:search query=sqlspec",
            "metadata_json": None,
            "inserted_at": datetime.now(UTC),
            "embedding": None,
        }
        await store.insert_memory_entries([record])
        results = await store.search_entries(query="sqlspec", app_name="docs", user_id="user_1")
        assert len(results) == 1
        return len(results)
    finally:
        await config.close_pool()


if __name__ == "__main__":
    print(asyncio.run(main()))