fast_api.py

# Copyright 2025 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.

from __future__ import annotations

import json
import logging
import os
from pathlib import Path
import shutil
from typing import Any
from typing import Mapping
from typing import Optional

import click
from fastapi import FastAPI
from fastapi import UploadFile
from fastapi.responses import FileResponse
from fastapi.responses import PlainTextResponse
from opentelemetry.sdk.trace import export
from opentelemetry.sdk.trace import TracerProvider
from starlette.types import Lifespan
from watchdog.observers import Observer

from ..artifacts.in_memory_artifact_service import InMemoryArtifactService
from ..auth.credential_service.in_memory_credential_service import InMemoryCredentialService
from ..evaluation.local_eval_set_results_manager import LocalEvalSetResultsManager
from ..evaluation.local_eval_sets_manager import LocalEvalSetsManager
from ..memory.in_memory_memory_service import InMemoryMemoryService
from ..runners import Runner
from ..sessions.in_memory_session_service import InMemorySessionService
from .adk_web_server import AdkWebServer
from .service_registry import get_service_registry
from .utils import envs
from .utils import evals
from .utils.agent_change_handler import AgentChangeEventHandler
from .utils.agent_loader import AgentLoader

logger = logging.getLogger("google_adk." + __name__)


def get_fast_api_app(
    *,
    agents_dir: str,
    session_service_uri: Optional[str] = None,
    session_db_kwargs: Optional[Mapping[str, Any]] = None,
    artifact_service_uri: Optional[str] = None,
    memory_service_uri: Optional[str] = None,
    eval_storage_uri: Optional[str] = None,
    allow_origins: Optional[list[str]] = None,
    web: bool,
    a2a: bool = False,
    host: str = "127.0.0.1",
    port: int = 8000,
    url_prefix: Optional[str] = None,
    trace_to_cloud: bool = False,
    otel_to_cloud: bool = False,
    reload_agents: bool = False,
    lifespan: Optional[Lifespan[FastAPI]] = None,
    extra_plugins: Optional[list[str]] = None,
    logo_text: Optional[str] = None,
    logo_image_url: Optional[str] = None,
) -> FastAPI:
  # Set up eval managers.
  if eval_storage_uri:
    gcs_eval_managers = evals.create_gcs_eval_managers_from_uri(
        eval_storage_uri
    )
    eval_sets_manager = gcs_eval_managers.eval_sets_manager
    eval_set_results_manager = gcs_eval_managers.eval_set_results_manager
  else:
    eval_sets_manager = LocalEvalSetsManager(agents_dir=agents_dir)
    eval_set_results_manager = LocalEvalSetResultsManager(agents_dir=agents_dir)

  service_registry = get_service_registry()

  # Build the Memory service
  if memory_service_uri:
    memory_service = service_registry.create_memory_service(
        memory_service_uri, agents_dir=agents_dir
    )
    if not memory_service:
      raise click.ClickException(
          "Unsupported memory service URI: %s" % memory_service_uri
      )
  else:
    memory_service = InMemoryMemoryService()

  # Build the Session service
  if session_service_uri:
    session_kwargs = session_db_kwargs or {}
    session_service = service_registry.create_session_service(
        session_service_uri, agents_dir=agents_dir, **session_kwargs
    )
    if not session_service:
      # Fallback to DatabaseSessionService if the service registry doesn't
      # support the session service URI scheme.
      from ..sessions.database_session_service import DatabaseSessionService

      session_service = DatabaseSessionService(
          db_url=session_service_uri, **session_kwargs
      )
  else:
    session_service = InMemorySessionService()

  # Build the Artifact service
  if artifact_service_uri:
    artifact_service = service_registry.create_artifact_service(
        artifact_service_uri, agents_dir=agents_dir
    )
    if not artifact_service:
      raise click.ClickException(
          "Unsupported artifact service URI: %s" % artifact_service_uri
      )
  else:
    artifact_service = InMemoryArtifactService()

  # Build  the Credential service
  credential_service = InMemoryCredentialService()

  # initialize Agent Loader
  agent_loader = AgentLoader(agents_dir)

  adk_web_server = AdkWebServer(
      agent_loader=agent_loader,
      session_service=session_service,
      artifact_service=artifact_service,
      memory_service=memory_service,
      credential_service=credential_service,
      eval_sets_manager=eval_sets_manager,
      eval_set_results_manager=eval_set_results_manager,
      agents_dir=agents_dir,
      extra_plugins=extra_plugins,
      logo_text=logo_text,
      logo_image_url=logo_image_url,
      url_prefix=url_prefix,
  )

  # Callbacks & other optional args for when constructing the FastAPI instance
  extra_fast_api_args = {}

  # TODO - Remove separate trace_to_cloud logic once otel_to_cloud stops being
  # EXPERIMENTAL.
  if trace_to_cloud and not otel_to_cloud:
    from opentelemetry.exporter.cloud_trace import CloudTraceSpanExporter

    def register_processors(provider: TracerProvider) -> None:
      envs.load_dotenv_for_agent("", agents_dir)
      if project_id := os.environ.get("GOOGLE_CLOUD_PROJECT", None):
        processor = export.BatchSpanProcessor(
            CloudTraceSpanExporter(project_id=project_id)
        )
        provider.add_span_processor(processor)
      else:
        logger.warning(
            "GOOGLE_CLOUD_PROJECT environment variable is not set. Tracing will"
            " not be enabled."
        )

    extra_fast_api_args.update(
        register_processors=register_processors,
    )

  if reload_agents:

    def setup_observer(observer: Observer, adk_web_server: AdkWebServer):
      agent_change_handler = AgentChangeEventHandler(
          agent_loader=agent_loader,
          runners_to_clean=adk_web_server.runners_to_clean,
          current_app_name_ref=adk_web_server.current_app_name_ref,
      )
      observer.schedule(agent_change_handler, agents_dir, recursive=True)
      observer.start()

    def tear_down_observer(observer: Observer, _: AdkWebServer):
      observer.stop()
      observer.join()

    extra_fast_api_args.update(
        setup_observer=setup_observer,
        tear_down_observer=tear_down_observer,
    )

  if web:
    BASE_DIR = Path(__file__).parent.resolve()
    ANGULAR_DIST_PATH = BASE_DIR / "browser"
    extra_fast_api_args.update(
        web_assets_dir=ANGULAR_DIST_PATH,
    )

  app = adk_web_server.get_fast_api_app(
      lifespan=lifespan,
      allow_origins=allow_origins,
      otel_to_cloud=otel_to_cloud,
      **extra_fast_api_args,
  )

  @app.post("/builder/save", response_model_exclude_none=True)
  async def builder_build(
      files: list[UploadFile], tmp: Optional[bool] = False
  ) -> bool:
    base_path = Path.cwd() / agents_dir
    for file in files:
      if not file.filename:
        logger.exception("Agent name is missing in the input files")
        return False
      agent_name, filename = file.filename.split("/")
      agent_dir = os.path.join(base_path, agent_name)
      try:
        # File name format: {app_name}/{agent_name}.yaml
        if tmp:
          agent_dir = os.path.join(agent_dir, "tmp/" + agent_name)
          os.makedirs(agent_dir, exist_ok=True)
          file_path = os.path.join(agent_dir, filename)
          with open(file_path, "wb") as buffer:
            shutil.copyfileobj(file.file, buffer)

        else:
          source_dir = os.path.join(agent_dir, "tmp/" + agent_name)
          destination_dir = agent_dir
          for item in os.listdir(source_dir):
            source_item = os.path.join(source_dir, item)
            destination_item = os.path.join(destination_dir, item)
            if os.path.isdir(source_item):
              shutil.copytree(source_item, destination_item, dirs_exist_ok=True)
            # Check if the item is a file
            elif os.path.isfile(source_item):
              shutil.copy2(source_item, destination_item)
      except Exception as e:
        logger.exception("Error in builder_build: %s", e)
        return False

    return True

  @app.post("/builder/app/{app_name}/cancel", response_model_exclude_none=True)
  async def builder_cancel(app_name: str) -> bool:
    base_path = Path.cwd() / agents_dir
    agent_dir = os.path.join(base_path, app_name)
    destination_dir = os.path.join(agent_dir, "tmp/" + app_name)
    source_dir = agent_dir
    source_items = set(os.listdir(source_dir))
    try:
      for item in os.listdir(destination_dir):
        if item in source_items:
          continue
        # If it doesn't exist in the source, delete it from the destination
        item_path = os.path.join(destination_dir, item)
        if os.path.isdir(item_path):
          shutil.rmtree(item_path)
        elif os.path.isfile(item_path):
          os.remove(item_path)

      for item in os.listdir(source_dir):
        source_item = os.path.join(source_dir, item)
        destination_item = os.path.join(destination_dir, item)
        if item == "tmp" and os.path.isdir(source_item):
          continue
        if os.path.isdir(source_item):
          shutil.copytree(source_item, destination_item, dirs_exist_ok=True)
        # Check if the item is a file
        elif os.path.isfile(source_item):
          shutil.copy2(source_item, destination_item)
    except Exception as e:
      logger.exception("Error in builder_build: %s", e)
      return False
    return True

  @app.get(
      "/builder/app/{app_name}",
      response_model_exclude_none=True,
      response_class=PlainTextResponse,
  )
  async def get_agent_builder(
      app_name: str,
      file_path: Optional[str] = None,
      tmp: Optional[bool] = False,
  ):
    base_path = Path.cwd() / agents_dir
    agent_dir = base_path / app_name
    if tmp:
      agent_dir = agent_dir / "tmp"
      agent_dir = agent_dir / app_name
    if not file_path:
      file_name = "root_agent.yaml"
      root_file_path = agent_dir / file_name
      if not root_file_path.is_file():
        return ""
      else:
        return FileResponse(
            path=root_file_path,
            media_type="application/x-yaml",
            filename="${app_name}.yaml",
            headers={"Cache-Control": "no-store"},
        )
    else:
      agent_file_path = agent_dir / file_path
      if not agent_file_path.is_file():
        return ""
      else:
        return FileResponse(
            path=agent_file_path,
            media_type="application/x-yaml",
            filename=file_path,
            headers={"Cache-Control": "no-store"},
        )

  if a2a:
    try:
      from a2a.server.apps import A2AStarletteApplication
      from a2a.server.request_handlers import DefaultRequestHandler
      from a2a.server.tasks import InMemoryTaskStore
      from a2a.types import AgentCard
      from a2a.utils.constants import AGENT_CARD_WELL_KNOWN_PATH

      from ..a2a.executor.a2a_agent_executor import A2aAgentExecutor

    except ImportError as e:
      import sys

      if sys.version_info < (3, 10):
        raise ImportError(
            "A2A requires Python 3.10 or above. Please upgrade your Python"
            " version."
        ) from e
      else:
        raise e
    # locate all a2a agent apps in the agents directory
    base_path = Path.cwd() / agents_dir
    # the root agents directory should be an existing folder
    if base_path.exists() and base_path.is_dir():
      a2a_task_store = InMemoryTaskStore()

      def create_a2a_runner_loader(captured_app_name: str):
        """Factory function to create A2A runner with proper closure."""

        async def _get_a2a_runner_async() -> Runner:
          return await adk_web_server.get_runner_async(captured_app_name)

        return _get_a2a_runner_async

      for p in base_path.iterdir():
        # only folders with an agent-card.json or agent.json file representing agent card are valid
        # a2a agents
        if p.is_file() or p.name.startswith(('.', '__pycache__')):
          continue

        json_file = p / 'agent-card.json'
        if not json_file.is_file():
          json_file = p / 'agent.json'
          if not json_file.is_file():
            continue

        app_name = p.name
        logger.info("Setting up A2A agent: %s", app_name)

        try:
          agent_executor = A2aAgentExecutor(
              runner=create_a2a_runner_loader(app_name),
          )

          request_handler = DefaultRequestHandler(
              agent_executor=agent_executor, task_store=a2a_task_store
          )
          with json_file.open("r", encoding="utf-8") as f:
            data = json.load(f)
            agent_card = AgentCard(**data)

          a2a_app = A2AStarletteApplication(
              agent_card=agent_card,
              http_handler=request_handler,
          )

          routes = a2a_app.routes(
              rpc_url=f"/a2a/{app_name}",
              agent_card_url=f"/a2a/{app_name}{AGENT_CARD_WELL_KNOWN_PATH}",
          )

          for new_route in routes:
            app.router.routes.append(new_route)

          logger.info("Successfully configured A2A agent: %s", app_name)

        except Exception as e:
          logger.error("Failed to setup A2A agent %s: %s", app_name, e)
          # Continue with other agents even if one fails

  return app