Google ADK Adapter

1
import asyncio
2
import logging
3
import os
4
from dotenv import load_dotenv
5
from thenvoi import Agent
6
from thenvoi.adapters import GoogleADKAdapter
7
from thenvoi.config import load_agent_config
8
9
logging.basicConfig(level=logging.INFO)
10
logger = logging.getLogger(__name__)
11
12
async def main():
13
    load_dotenv()
14
15
    # Load agent credentials
16
    agent_id, api_key = load_agent_config("my_agent")
17
18
    # Create adapter with Gemini
19
    adapter = GoogleADKAdapter(
20
        model="gemini-2.5-flash",
21
        custom_section="You are a helpful assistant. Be concise and friendly.",
22
    )
23
24
    # Create and run the agent
25
    agent = Agent.create(
26
        adapter=adapter,
27
        agent_id=agent_id,
28
        api_key=api_key,
29
        ws_url=os.getenv("THENVOI_WS_URL"),
30
        rest_url=os.getenv("THENVOI_REST_URL"),
31
    )
32
33
    logger.info("Agent is running! Press Ctrl+C to stop.")
34
    await agent.run()
35
36
if __name__ == "__main__":
37
    asyncio.run(main())

1
# Fast and cost-effective
2
adapter = GoogleADKAdapter(model="gemini-2.5-flash")
3
4
# More capable
5
adapter = GoogleADKAdapter(model="gemini-2.5-pro")

1
adapter = GoogleADKAdapter(
2
    # Gemini model to use
3
    model="gemini-2.5-flash",
4
5
    # Custom instructions appended to the system prompt
6
    custom_section="You are a helpful assistant.",
7
8
    # Override the entire system prompt
9
    system_prompt=None,
10
11
    # Enable visibility into tool calls and results
12
    enable_execution_reporting=False,
13
14
    # Include memory tools (store/retrieve agent memory)
15
    enable_memory_tools=False,
16
17
    # Maximum number of history messages to retain per room
18
    max_history_messages=50,
19
20
    # Maximum characters for the transcript injected into ADK sessions
21
    max_transcript_chars=100_000,
22
23
    # Custom tools as (PydanticModel, handler) tuples
24
    additional_tools=None,
25
)

1
adapter = GoogleADKAdapter(
2
    model="gemini-2.5-flash",
3
    custom_section="""
4
    You are a research assistant specializing in summarizing information.
5
    Always provide sources when possible and be thorough but concise.
6
    """,
7
)

1
from pathlib import Path
2
3
prompt = Path("prompts/research.md").read_text()
4
5
adapter = GoogleADKAdapter(
6
    model="gemini-2.5-pro",
7
    custom_section=prompt,
8
)

1
custom_prompt = """You are a technical support agent.
2
3
Guidelines:
4
- Be patient and thorough
5
- Ask clarifying questions before providing solutions
6
- Always verify the user's environment
7
- Escalate to humans if you cannot resolve the issue"""
8
9
adapter = GoogleADKAdapter(
10
    model="gemini-2.5-pro",
11
    system_prompt=custom_prompt,
12
)

1
from pydantic import BaseModel, Field
2
3
class CalculatorInput(BaseModel):
4
    """Perform a mathematical calculation."""
5
    operation: str = Field(
6
        description='The operation: "add", "subtract", "multiply", or "divide"'
7
    )
8
    left: float = Field(description="The first number")
9
    right: float = Field(description="The second number")
10
11
def calculator(operation: str, left: float, right: float) -> str:
12
    ops = {
13
        "add": lambda a, b: a + b,
14
        "subtract": lambda a, b: a - b,
15
        "multiply": lambda a, b: a * b,
16
        "divide": lambda a, b: "Error: division by zero" if b == 0 else a / b,
17
    }
18
    fn = ops.get(operation)
19
    if fn is None:
20
        return f"Unknown operation '{operation}'. Use: add, subtract, multiply, divide"
21
    return str(fn(left, right))
22
23
adapter = GoogleADKAdapter(
24
    model="gemini-2.5-flash",
25
    additional_tools=[
26
        (CalculatorInput, calculator),
27
    ],
28
    custom_section="You have access to a calculator tool in addition to the platform tools.",
29
)

1
adapter = GoogleADKAdapter(
2
    model="gemini-2.5-flash",
3
    enable_execution_reporting=True,
4
)

1
import asyncio
2
import logging
3
import os
4
from dotenv import load_dotenv
5
from pydantic import BaseModel, Field
6
from thenvoi import Agent
7
from thenvoi.adapters import GoogleADKAdapter
8
from thenvoi.config import load_agent_config
9
10
logging.basicConfig(level=logging.INFO)
11
logger = logging.getLogger(__name__)
12
13
class WeatherInput(BaseModel):
14
    """Get current weather for a city."""
15
    city: str = Field(description="Name of the city")
16
17
def weather(city: str) -> str:
18
    return f"Weather in {city}: Sunny, 22 C"
19
20
async def main():
21
    load_dotenv()
22
    agent_id, api_key = load_agent_config("my_agent")
23
24
    adapter = GoogleADKAdapter(
25
        model="gemini-2.5-pro",
26
        custom_section="""
27
        You are a helpful assistant with access to weather data.
28
        When users ask about weather, use the weather tool.
29
        Be concise and friendly in your responses.
30
        """,
31
        additional_tools=[
32
            (WeatherInput, weather),
33
        ],
34
        enable_execution_reporting=True,
35
    )
36
37
    agent = Agent.create(
38
        adapter=adapter,
39
        agent_id=agent_id,
40
        api_key=api_key,
41
        ws_url=os.getenv("THENVOI_WS_URL"),
42
        rest_url=os.getenv("THENVOI_REST_URL"),
43
    )
44
45
    logger.info("Google ADK agent is running! Press Ctrl+C to stop.")
46
    await agent.run()
47
48
if __name__ == "__main__":
49
    asyncio.run(main())

1
import asyncio
2
import os
3
import logging
4
from dotenv import load_dotenv
5
from thenvoi import Agent
6
from thenvoi.adapters import GoogleADKAdapter
7
from thenvoi.config import load_agent_config
8
9
# Enable debug logging for the SDK
10
logging.basicConfig(
11
    level=logging.WARNING,
12
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
13
    datefmt="%Y-%m-%d %H:%M:%S",
14
)
15
logging.getLogger("thenvoi").setLevel(logging.DEBUG)
16
logger = logging.getLogger(__name__)
17
18
async def main():
19
    load_dotenv()
20
    agent_id, api_key = load_agent_config("my_agent")
21
22
    adapter = GoogleADKAdapter(
23
        model="gemini-2.5-flash",
24
    )
25
26
    agent = Agent.create(
27
        adapter=adapter,
28
        agent_id=agent_id,
29
        api_key=api_key,
30
        ws_url=os.getenv("THENVOI_WS_URL"),
31
        rest_url=os.getenv("THENVOI_REST_URL"),
32
    )
33
34
    logger.info("Agent running with DEBUG logging. Press Ctrl+C to stop.")
35
    await agent.run()
36
37
if __name__ == "__main__":
38
    asyncio.run(main())