Agent Development Kit

1
import datetime
2
from zoneinfo import ZoneInfo
3
4
from google.adk.agents import Agent
5
from opik.integrations.adk import OpikTracer
6
7
8
def get_weather(city: str) -> dict:
9
    if city.lower() == "new york":
10
        return {
11
            "status": "success",
12
            "report": (
13
                "The weather in New York is sunny with a temperature of 25 degrees"
14
                " Celsius (41 degrees Fahrenheit)."
15
            ),
16
        }
17
    else:
18
        return {
19
            "status": "error",
20
            "error_message": f"Weather information for '{city}' is not available.",
21
        }
22
23
24
def get_current_time(city: str) -> dict:
25
    if city.lower() == "new york":
26
        tz_identifier = "America/New_York"
27
    else:
28
        return {
29
            "status": "error",
30
            "error_message": (f"Sorry, I don't have timezone information for {city}."),
31
        }
32
33
    tz = ZoneInfo(tz_identifier)
34
    now = datetime.datetime.now(tz)
35
    report = f'The current time in {city} is {now.strftime("%Y-%m-%d %H:%M:%S %Z%z")}'
36
    return {"status": "success", "report": report}
37
38
39
opik_tracer = OpikTracer()
40
41
root_agent = Agent(
42
    name="weather_time_agent",
43
    model="gemini-2.0-flash-exp",
44
    description=("Agent to answer questions about the time and weather in a city."),
45
    instruction=("I can answer your questions about the time and weather in a city."),
46
    tools=[get_weather, get_current_time],
47
    before_agent_callback=opik_tracer.before_agent_callback,
48
    after_agent_callback=opik_tracer.after_agent_callback,
49
    before_model_callback=opik_tracer.before_model_callback,
50
    after_model_callback=opik_tracer.after_model_callback,
51
    before_tool_callback=opik_tracer.before_tool_callback,
52
    after_tool_callback=opik_tracer.after_tool_callback,
53
)

1
async for event in runner.run_async(user_id=user_id, session_id=session_id, new_message=content):
2
    if event.is_final_response():
3
        ...
4
        break  # Stop processing events once the final response is found

1
from opik.integrations.adk import OpikTracer, track_adk_agent_recursive
2
opik_tracer = OpikTracer()
3
4
translator_to_english = adk_agents.Agent(
5
    name="Translator",
6
    model=MODEL_NAME,
7
    description="Translates text to English.",
8
)
9
summarizer = adk_agents.Agent(
10
    name="Summarizer",
11
    model=MODEL_NAME,
12
    description="Summarizes text to 1 sentence.",
13
)
14
root_agent = adk_agents.SequentialAgent(
15
    name="TextProcessingAssistant",
16
    sub_agents=[translator_to_english, summarizer],
17
    description="Runs translator to english then summarizer, in order.",
18
)
19
20
track_adk_agent_recursive(root_agent, opik_tracer)