Building agents
with Google
ADK 2.0

from google.adk.apps import App
from google.adk.runners import InMemoryRunner
from google.genai import types

app = App(name="time_app", root_agent=root_agent)
runner = InMemoryRunner(app=app)
session = await runner.session_service.create_session(
    app_name="time_app", user_id="u1", state={"user:name": "Ada"})
msg = types.Content(role="user",
                    parts=[types.Part(text="Time in Paris?")])
async for event in runner.run_async(
        user_id="u1", session_id=session.id, new_message=msg):
    for call in event.get_function_calls():
        print("TOOL CALL", call.name, call.args)
    if event.is_final_response() and event.content:
        print(event.author, ":", event.content.parts[0].text)

class CountInvocationPlugin(BasePlugin):
  def __init__(self) -> None:
    super().__init__(name="count_invocation")
    self.agent_count = 0
    self.llm_request_count = 0
  async def before_agent_callback(
      self, *, agent, callback_context) -> None:
    self.agent_count += 1
  async def before_model_callback(
      self, *, callback_context, llm_request) -> None:
    self.llm_request_count += 1   # Observe: return None

app = App(name="test_app_with_plugin",
          root_agent=root_agent,
          plugins=[CountInvocationPlugin()])
runner = InMemoryRunner(app=app)

from google.adk.events import RequestInput
def step1():  # the reply goes to the next node
  yield RequestInput(message="Enter a number:")

def handle_human_review(node_input: str):
  if node_input == "reject":
    yield Event(route="rejected")
  elif node_input == "approve":
    yield Event(route="approved")
  else:
    yield Event(state={"feedback": node_input}, route="revise")

root_agent = Workflow(name="request_input", edges=[
  ("START", process_input, draft_email,
   request_human_review, handle_human_review),
  (handle_human_review, {"revise": draft_email,
    "approved": send_email, "rejected": reject_email})])

from google.adk.workflow import RetryConfig, node
@node(retry_config=RetryConfig(max_attempts=5, initial_delay=1))
def get_weather(ctx: Context) -> str:
  yield Event(message=f"attempt {ctx.attempt_count}")
  if random.random() < 0.7:
    raise HTTPError(url="http://mock-api.example.com",
                    code=500, msg="ISE", hdrs={}, fp=None)
  yield "sunny"

try:
    data = call_api()
except Exception:          # masks the failure: retries never fire
    return {"status": "error"}
except BaseException: ...  # traps NodeInterruptedError: breaks HITL

app = App(name="my_resumable_agent", root_agent=root_agent,
    resumability_config=ResumabilityConfig(is_resumable=True))
# resume: the same user, session and invocation id
async for event in runner.run_async(user_id="u_123",
        session_id="s_abc", invocation_id="invocation-123"):
    print(event)

from google.adk.tools import FunctionTool, ToolContext

def add_to_cart(item: str, tool_context: ToolContext) -> dict:
    """Adds an item to the user's cart."""
    cart = tool_context.state.get("user:cart", [])
    cart.append(item)
    tool_context.state["user:cart"] = cart      # all sessions
    tool_context.state["temp:last_op"] = "add"  # this call only
    return {"status": "success", "cart_size": len(cart)}

async def confirmation_threshold(amount: int,
                                 tool_context: ToolContext) -> bool:
    return amount > 1000
tools = [FunctionTool(reimburse, require_confirmation=True),
         FunctionTool(reimburse_big,
                      require_confirmation=confirmation_threshold)]

from google.adk.tools.mcp_tool import (
    McpToolset, StreamableHTTPConnectionParams)
root_agent = LlmAgent(
    model=MODEL, name="currency_agent",
    tools=[McpToolset(connection_params=
        StreamableHTTPConnectionParams(url=MCP_SERVER_URL))])

from google.adk.a2a.utils.agent_to_a2a import to_a2a
a2a_app = to_a2a(root_agent, host=host, port=port,
                 protocol=protocol)   # ASGI app + Agent Card

from google.adk.agents.remote_a2a_agent import (
    RemoteA2aAgent, AGENT_CARD_WELL_KNOWN_PATH)
currency_remote_agent = RemoteA2aAgent(name="currency_agent",
    agent_card=f"{CURRENCY_AGENT_URL}{AGENT_CARD_WELL_KNOWN_PATH}")

from google.adk import Agent
from google.adk.tools import ToolContext, load_artifacts
from google.genai import Client, types

async def generate_image(prompt: str, tool_context: ToolContext):
  """Generates an image based on the prompt."""
  response = Client().models.generate_images(
      model="imagen-3.0-generate-002",  # retired model ID
      prompt=prompt, config={"number_of_images": 1})
  image_bytes = response.generated_images[0].image.image_bytes
  part = types.Part.from_bytes(data=image_bytes,
                               mime_type="image/png")
  await tool_context.save_artifact("image.png", part)  # -> version 0
  return {"status": "success", "filename": "image.png"}

root_agent = Agent(name="root_agent",
                   tools=[generate_image, load_artifacts])