adk-python/contributing/samples/adk_triaging_agent/main.py

# Copyright 2026 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import asyncio
import time

from adk_triaging_agent import agent
from adk_triaging_agent.agent import LABEL_TO_OWNER
from adk_triaging_agent.settings import EVENT_NAME
from adk_triaging_agent.settings import GITHUB_BASE_URL
from adk_triaging_agent.settings import ISSUE_BODY
from adk_triaging_agent.settings import ISSUE_COUNT_TO_PROCESS
from adk_triaging_agent.settings import ISSUE_NUMBER
from adk_triaging_agent.settings import ISSUE_TITLE
from adk_triaging_agent.settings import OWNER
from adk_triaging_agent.settings import REPO
from adk_triaging_agent.utils import get_request
from adk_triaging_agent.utils import parse_number_string
from google.adk.agents.run_config import RunConfig
from google.adk.runners import InMemoryRunner
from google.adk.runners import Runner
from google.genai import types
import requests

APP_NAME = "adk_triage_app"
USER_ID = "adk_triage_user"


async def fetch_specific_issue_details(issue_number: int):
  """Fetches details for a single issue if it needs triaging."""
  url = f"{GITHUB_BASE_URL}/repos/{OWNER}/{REPO}/issues/{issue_number}"
  print(f"Fetching details for specific issue: {url}")

  try:
    issue_data = get_request(url)
    labels = issue_data.get("labels", [])
    label_names = {label["name"] for label in labels}
    assignees = issue_data.get("assignees", [])

    # Check issue state
    component_labels = set(LABEL_TO_OWNER.keys())
    has_planned = "planned" in label_names
    existing_component_labels = label_names & component_labels
    has_component = bool(existing_component_labels)
    has_assignee = len(assignees) > 0

    # Determine what actions are needed
    needs_component_label = not has_component
    needs_owner = has_planned and not has_assignee

    if needs_component_label or needs_owner:
      print(
          f"Issue #{issue_number} needs triaging. "
          f"needs_component_label={needs_component_label}, "
          f"needs_owner={needs_owner}"
      )
      return {
          "number": issue_data["number"],
          "title": issue_data["title"],
          "body": issue_data.get("body", ""),
          "has_planned_label": has_planned,
          "has_component_label": has_component,
          "existing_component_label": (
              list(existing_component_labels)[0]
              if existing_component_labels
              else None
          ),
          "needs_component_label": needs_component_label,
          "needs_owner": needs_owner,
      }
    else:
      print(f"Issue #{issue_number} is already fully triaged. Skipping.")
      return None
  except requests.exceptions.RequestException as e:
    print(f"Error fetching issue #{issue_number}: {e}")
    if hasattr(e, "response") and e.response is not None:
      print(f"Response content: {e.response.text}")
    return None


async def call_agent_async(
    runner: Runner, user_id: str, session_id: str, prompt: str
) -> str:
  """Call the agent asynchronously with the user's prompt."""
  content = types.Content(
      role="user", parts=[types.Part.from_text(text=prompt)]
  )

  final_response_text = ""
  async for event in runner.run_async(
      user_id=user_id,
      session_id=session_id,
      new_message=content,
      run_config=RunConfig(save_input_blobs_as_artifacts=False),
  ):
    if (
        event.content
        and event.content.parts
        and hasattr(event.content.parts[0], "text")
        and event.content.parts[0].text
    ):
      print(f"** {event.author} (ADK): {event.content.parts[0].text}")
      if event.author == agent.root_agent.name:
        final_response_text += event.content.parts[0].text

  return final_response_text


async def main():
  runner = InMemoryRunner(
      agent=agent.root_agent,
      app_name=APP_NAME,
  )
  session = await runner.session_service.create_session(
      user_id=USER_ID,
      app_name=APP_NAME,
  )

  if EVENT_NAME == "issues" and ISSUE_NUMBER:
    print(f"EVENT: Processing specific issue due to '{EVENT_NAME}' event.")
    issue_number = parse_number_string(ISSUE_NUMBER)
    if not issue_number:
      print(f"Error: Invalid issue number received: {ISSUE_NUMBER}.")
      return

    specific_issue = await fetch_specific_issue_details(issue_number)
    if specific_issue is None:
      print(
          f"No issue details found for #{issue_number} that needs triaging,"
          " or an error occurred. Skipping agent interaction."
      )
      return

    issue_title = ISSUE_TITLE or specific_issue["title"]
    issue_body = ISSUE_BODY or specific_issue["body"]
    needs_component_label = specific_issue.get("needs_component_label", True)
    needs_owner = specific_issue.get("needs_owner", False)
    existing_component_label = specific_issue.get("existing_component_label")

    prompt = (
        f"Triage GitHub issue #{issue_number}.\n\n"
        f'Title: "{issue_title}"\n'
        f'Body: "{issue_body}"\n\n'
        f"Issue state: needs_component_label={needs_component_label}, "
        f"needs_owner={needs_owner}, "
        f"existing_component_label={existing_component_label}"
    )
  else:
    print(f"EVENT: Processing batch of issues (event: {EVENT_NAME}).")
    issue_count = parse_number_string(ISSUE_COUNT_TO_PROCESS, default_value=3)
    prompt = (
        f"Please use 'list_untriaged_issues' to find {issue_count} issues that"
        " need triaging, then triage each one according to your instructions."
    )

  response = await call_agent_async(runner, USER_ID, session.id, prompt)
  print(f"<<<< Agent Final Output: {response}\n")


if __name__ == "__main__":
  start_time = time.time()
  print(
      f"Start triaging {OWNER}/{REPO} issues at"
      f" {time.strftime('%Y-%m-%d %H:%M:%S', time.gmtime(start_time))}"
  )
  print("-" * 80)
  asyncio.run(main())
  print("-" * 80)
  end_time = time.time()
  print(
      "Triaging finished at"
      f" {time.strftime('%Y-%m-%d %H:%M:%S', time.gmtime(end_time))}",
  )
  print("Total script execution time:", f"{end_time - start_time:.2f} seconds")