Metadata-Version: 2.4
Name: dbos-google-adk
Version: 0.1.0
Summary: DBOS integration for the Google Agent Development Kit
Project-URL: Homepage, https://github.com/dbos-inc/dbos-google-adk
Project-URL: Repository, https://github.com/dbos-inc/dbos-google-adk
Author: DBOS, Inc.
License: MIT
License-File: LICENSE
Keywords: adk,agents,dbos,durable,google,workflow
Classifier: Development Status :: 4 - Beta
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Requires-Python: >=3.10
Requires-Dist: dbos>=2.21.0
Requires-Dist: google-adk>=1.32.0
Requires-Dist: mypy>=2.0.0
Description-Content-Type: text/markdown

# Durable Google ADK Agents

Durable execution for the [Google Agent Development Kit](https://github.com/google/adk-python) using [DBOS](https://github.com/dbos-inc/dbos-transact-py).

## Installation

```bash
pip install dbos-google-adk
```

## Usage

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

```python
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:

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

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