dbos-google-adk 0.1.0

DBOS integration for the Google Agent Development Kit

Durable Google ADK Agents

Durable execution for the Google Agent Development Kit using DBOS.

Installation

pip install dbos-google-adk

Usage

Add DBOSPlugin to your Runner and call it from a @DBOS.workflow(). Annotate tool calls with @DBOS.step().

import asyncio
import logging

from dbos import DBOS, DBOSConfig
from dbos_google_adk import DBOSPlugin
from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

# Suppress cosmetic OpenTelemetry warning. See google/adk-python#4894.
logging.getLogger("opentelemetry.context").setLevel(logging.CRITICAL)


# Decorate tool calls with @DBOS.step() for durable execution
@DBOS.step()
async def get_weather(city: str) -> str:
    """Get the weather for a city."""
    return f"Sunny in {city}"


agent = LlmAgent(name="weather", model="gemini-2.5-flash", tools=[get_weather])
runner = Runner(
    app_name="my-agent",
    agent=agent,
    plugins=[DBOSPlugin()],
    session_service=InMemorySessionService(),
)


# Drive the agent from a DBOS workflow for durable execution
@DBOS.workflow()
async def run_agent(user_id: str, session_id: str, message: str) -> str:
    new_message = types.Content(role="user", parts=[types.Part.from_text(text=message)])
    async for event in runner.run_async(
        user_id=user_id, session_id=session_id, new_message=new_message
    ):
        if event.is_final_response():
            return event.content.parts[0].text
    return ""


async def main():
    config: DBOSConfig = {"name": "my-agent"}
    DBOS(config=config)
    DBOS.launch()

    await runner.session_service.create_session(
        app_name="my-agent", user_id="u", session_id="s"
    )
    print(await run_agent("u", "s", "How is the weather in San Francisco?"))


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

DBOSPlugin checkpoints every model call through DBOS so that an interrupted workflow resumes from the last completed step on retry. It must be used from within a @DBOS.workflow().

For durable event compaction, wrap your summarizer with DBOSEventSummarizer so the compaction LLM call is also checkpointed:

from dbos_google_adk import DBOSEventSummarizer
from google.adk.models.google_llm import Gemini

summarizer = DBOSEventSummarizer.from_llm(Gemini(model="gemini-2.5-flash"))