feat: Adds adk web options for custom logo

Allows users to configure a custom text and logo for their ADK Web app using `--logo-text` and `--logo-image-url` flags.

PiperOrigin-RevId: 814016542

Author: hoonji

src/google/adk/cli/adk_web_server.py

@@ -17,6 +17,7 @@
 import asyncio
 from contextlib import asynccontextmanager
 import importlib
+import json
 import logging
 import os
 import time
@@ -395,6 +396,9 @@ class AdkWebServer:
         managing evaluation set results.
       agents_dir: Root directory containing subdirs for agents with those
         containing resources (e.g. .env files, eval sets, etc.) for the agents.
+      extra_plugins: A list of fully qualified names of extra plugins to load.
+      logo_text: Text to display in the logo of the UI.
+      logo_image_url: URL of an image to display as logo of the UI.
       runners_to_clean: Set of runner names marked for cleanup.
       current_app_name_ref: A shared reference to the latest ran app name.
       runner_dict: A dict of instantiated runners for each app.
@@ -412,6 +416,8 @@ def __init__(
       eval_set_results_manager: EvalSetResultsManager,
       agents_dir: str,
       extra_plugins: Optional[list[str]] = None,
+      logo_text: Optional[str] = None,
+      logo_image_url: Optional[str] = None,
   ):
     self.agent_loader = agent_loader
     self.session_service = session_service
@@ -422,6 +428,8 @@ def __init__(
     self.eval_set_results_manager = eval_set_results_manager
     self.agents_dir = agents_dir
     self.extra_plugins = extra_plugins or []
+    self.logo_text = logo_text
+    self.logo_image_url = logo_image_url
     # Internal propeties we want to allow being modified from callbacks.
     self.runners_to_clean: set[str] = set()
     self.current_app_name_ref: SharedValue[str] = SharedValue(value="")
@@ -506,6 +514,52 @@ def _import_plugin_object(self, qualified_name: str) -> Any:
     module = importlib.import_module(module_name)
     return getattr(module, obj_name)
 
+  def _setup_runtime_config(self, web_assets_dir: str):
+    """Sets up the runtime config for the web server."""
+    # Read existing runtime config file.
+    runtime_config_path = os.path.join(
+        web_assets_dir, "assets", "config", "runtime-config.json"
+    )
+    runtime_config = {}
+    try:
+      with open(runtime_config_path, "r") as f:
+        runtime_config = json.load(f)
+    except FileNotFoundError:
+      logger.info(
+          "File not found: %s. A new runtime config file will be created.",
+          runtime_config_path,
+      )
+    except json.JSONDecodeError:
+      logger.warning(
+          "Failed to decode JSON from %s. The file content will be"
+          " overwritten.",
+          runtime_config_path,
+      )
+
+    # Set custom logo config.
+    if self.logo_text or self.logo_image_url:
+      if not self.logo_text or not self.logo_image_url:
+        raise ValueError(
+            "Both --logo-text and --logo-image-url must be defined when using"
+            " logo config."
+        )
+      runtime_config["logo"] = {
+          "text": self.logo_text,
+          "imageUrl": self.logo_image_url,
+      }
+    elif "logo" in runtime_config:
+      del runtime_config["logo"]
+
+    # Write the runtime config file.
+    try:
+      os.makedirs(os.path.dirname(runtime_config_path), exist_ok=True)
+      with open(runtime_config_path, "w") as f:
+        json.dump(runtime_config, f, indent=2)
+    except IOError as e:
+      logger.error(
+          "Failed to write runtime config file %s: %s", runtime_config_path, e
+      )
+
   def get_fast_api_app(
       self,
       lifespan: Optional[Lifespan[FastAPI]] = None,
@@ -570,6 +624,8 @@ async def internal_lifespan(app: FastAPI):
             export_lib.SimpleSpanProcessor(memory_exporter),
         ],
     )
+    if web_assets_dir:
+      self._setup_runtime_config(web_assets_dir)
 
     # TODO - register_processors to be removed once --otel_to_cloud is no
     # longer experimental.
@@ -1413,6 +1469,13 @@ async def process_messages():
       mimetypes.add_type("application/javascript", ".js", True)
       mimetypes.add_type("text/javascript", ".js", True)
 
+      @app.get("/dev-ui/config")
+      async def get_ui_config():
+        return {
+            "logo_text": self.logo_text,
+            "logo_image_url": self.logo_image_url,
+        }
+
       @app.get("/")
       async def redirect_root_to_dev_ui():
         return RedirectResponse("/dev-ui/")

src/google/adk/cli/cli_tools_click.py

@@ -677,6 +677,34 @@ def cli_eval(
       pretty_print_eval_result(eval_result)
 
 
+def web_options():
+  """Decorator to add web UI options to click commands."""
+
+  def decorator(func):
+    @click.option(
+        "--logo-text",
+        type=str,
+        help="Optional. The text to display in the logo of the web UI.",
+        default=None,
+    )
+    @click.option(
+        "--logo-image-url",
+        type=str,
+        help=(
+            "Optional. The URL of the image to display in the logo of the"
+            " web UI."
+        ),
+        default=None,
+    )
+    @functools.wraps(func)
+    def wrapper(*args, **kwargs):
+      return func(*args, **kwargs)
+
+    return wrapper
+
+  return decorator
+
+
 def adk_services_options():
   """Decorator to add ADK services options to click commands."""
 
@@ -872,6 +900,7 @@ def wrapper(ctx, *args, **kwargs):
 
 @main.command("web")
 @fast_api_common_options()
+@web_options()
 @adk_services_options()
 @deprecated_adk_services_options()
 @click.argument(
@@ -899,6 +928,8 @@ def cli_web(
     a2a: bool = False,
     reload_agents: bool = False,
     extra_plugins: Optional[list[str]] = None,
+    logo_text: Optional[str] = None,
+    logo_image_url: Optional[str] = None,
 ):
   """Starts a FastAPI server with Web UI for agents.
 
@@ -951,6 +982,8 @@ async def _lifespan(app: FastAPI):
       port=port,
       reload_agents=reload_agents,
       extra_plugins=extra_plugins,
+      logo_text=logo_text,
+      logo_image_url=logo_image_url,
   )
   config = uvicorn.Config(
       app,

src/google/adk/cli/fast_api.py

@@ -71,6 +71,8 @@ def get_fast_api_app(
     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:
@@ -189,6 +191,8 @@ def _parse_agent_engine_resource_name(agent_engine_id_or_resource_name):
       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,
   )
 
   # Callbacks & other optional args for when constructing the FastAPI instance