ADK Session Service (AioSQLite)

Persist Google ADK sessions using SQLSpec’s AioSQLite store.

uv run python docs/examples/extensions/adk/basic_aiosqlite.py

Source

 1"""Persist Google ADK sessions with SQLSpec + AioSQLite."""
 2
 3import asyncio
 4from datetime import datetime, timezone
 5
 6from google.adk.events.event import Event
 7from google.genai import types
 8
 9from sqlspec.adapters.aiosqlite import AiosqliteConfig
10from sqlspec.adapters.aiosqlite.adk import AiosqliteADKStore
11from sqlspec.extensions.adk import SQLSpecSessionService
12
13__all__ = ("main",)
14
15
16def _event(author: str, text: str) -> Event:
17    return Event(
18        id=f"evt_{author}",
19        invocation_id="inv_1",
20        author=author,
21        branch="main",
22        actions=[],
23        timestamp=datetime.now(timezone.utc).timestamp(),
24        content=types.Content(parts=[types.Part(text=text)]),
25        partial=False,
26        turn_complete=True,
27    )
28
29
30async def main() -> None:
31    """Create a session, append two events, and read the transcript."""
32    config = AiosqliteConfig(pool_config={"database": ":memory:"})
33    store = AiosqliteADKStore(config)
34    await store.create_tables()
35    service = SQLSpecSessionService(store)
36    session = await service.create_session(app_name="docs", user_id="demo", state={"mode": "chat"})
37    await service.append_event(session, _event("user", "How does SQLSpec store sessions?"))
38    await service.append_event(session, _event("assistant", "Sessions live in SQLite tables via the ADK store."))
39    replay = await service.get_session(app_name="docs", user_id="demo", session_id=session.id)
40    total = len(replay.events) if replay else 0
41    print({"session": session.id, "events": total})
42
43
44if __name__ == "__main__":
45    asyncio.run(main())

 1"""Persist Google ADK sessions with SQLSpec + AioSQLite."""
 2
 3import asyncio
 4from datetime import datetime, UTC
 5
 6from google.adk.events.event import Event
 7from google.genai import types
 8
 9from sqlspec.adapters.aiosqlite import AiosqliteConfig
10from sqlspec.adapters.aiosqlite.adk import AiosqliteADKStore
11from sqlspec.extensions.adk import SQLSpecSessionService
12
13__all__ = ("main",)
14
15
16def _event(author: str, text: str) -> Event:
17    return Event(
18        id=f"evt_{author}",
19        invocation_id="inv_1",
20        author=author,
21        branch="main",
22        actions=[],
23        timestamp=datetime.now(UTC).timestamp(),
24        content=types.Content(parts=[types.Part(text=text)]),
25        partial=False,
26        turn_complete=True,
27    )
28
29
30async def main() -> None:
31    """Create a session, append two events, and read the transcript."""
32    config = AiosqliteConfig(pool_config={"database": ":memory:"})
33    store = AiosqliteADKStore(config)
34    await store.create_tables()
35    service = SQLSpecSessionService(store)
36    session = await service.create_session(app_name="docs", user_id="demo", state={"mode": "chat"})
37    await service.append_event(session, _event("user", "How does SQLSpec store sessions?"))
38    await service.append_event(session, _event("assistant", "Sessions live in SQLite tables via the ADK store."))
39    replay = await service.get_session(app_name="docs", user_id="demo", session_id=session.id)
40    total = len(replay.events) if replay else 0
41    print({"session": session.id, "events": total})
42
43
44if __name__ == "__main__":
45    asyncio.run(main())