Add avatar configuration to server commands

src/google/adk/cli/api_server.py

@@ -839,6 +839,7 @@ def __init__(
       auto_create_session: bool = False,
       trigger_sources: Optional[list[str]] = None,
       default_llm_model: Optional[str] = None,
+      avatar_config: Optional[types.AvatarConfig] = None,
   ):
     self.agent_loader = agent_loader
     self.session_service = session_service
@@ -859,6 +860,7 @@ def __init__(
     self.auto_create_session = auto_create_session
     self.trigger_sources = trigger_sources
     self.default_llm_model = default_llm_model
+    self.avatar_config = avatar_config
     self.default_app_name = os.getenv("ADK_DEFAULT_APP_NAME")
 
   async def get_runner_async(self, app_name: str) -> Runner:
@@ -2061,6 +2063,7 @@ async def forward_events():
             ),
             save_live_blob=save_live_blob,
             explicit_vad_signal=explicit_vad_signal,
+            avatar_config=self.avatar_config,
         )
         async with Aclosing(
             runner.run_live(

src/google/adk/cli/cli_tools_click.py

@@ -49,6 +49,7 @@
 
 if TYPE_CHECKING:
   from fastapi import FastAPI
+  from google.genai import types
 
   from ..agents.llm_agent import LlmAgent
 
@@ -85,6 +86,37 @@ def _parse_streaming_mode(
   return mode
 
 
+def _parse_avatar_config(
+    _ctx: click.Context,
+    param: click.Parameter,
+    value: str | None,
+) -> types.AvatarConfig | None:
+  """Parses an inline JSON object or JSON file into an avatar config."""
+  if value is None:
+    return None
+
+  if value.lstrip().startswith("{"):
+    config_json = value
+  else:
+    try:
+      config_json = Path(value).read_text(encoding="utf-8")
+    except OSError as exc:
+      raise click.BadParameter(
+          f"could not read avatar configuration file {value!r}: {exc}",
+          param=param,
+      ) from exc
+
+  from google.genai import types
+
+  try:
+    return types.AvatarConfig.model_validate_json(config_json)
+  except ValueError as exc:
+    raise click.BadParameter(
+        f"avatar configuration must be a valid AvatarConfig JSON object: {exc}",
+        param=param,
+    ) from exc
+
+
 def _logging_options():
   """Decorator to add logging options to click commands."""
 
@@ -1927,6 +1959,16 @@ def decorator(func):
         ),
         default=None,
     )
+    @click.option(
+        "--avatar_config",
+        type=str,
+        callback=_parse_avatar_config,
+        help=(
+            "Optional. AvatarConfig as an inline JSON object or a path to a"
+            " JSON file. Applied to live sessions."
+        ),
+        default=None,
+    )
     # Parsed into list[str] by the wrapper below (server commands need a list).
     @click.option(
         "--trigger_sources",
@@ -2012,6 +2054,7 @@ def cli_web(
     logo_text: str | None = None,
     logo_image_url: str | None = None,
     trigger_sources: list[str] | None = None,
+    avatar_config: types.AvatarConfig | None = None,
 ):
   """Starts a FastAPI server with Web UI for agents.
 
@@ -2082,6 +2125,7 @@ async def _lifespan(app: FastAPI):
       logo_image_url=logo_image_url,
       trigger_sources=trigger_sources,
       default_llm_model=default_llm_model,
+      avatar_config=avatar_config,
   )
   config = uvicorn.Config(
       app,
@@ -2163,6 +2207,7 @@ def cli_api_server(
     with_ui: bool = False,
     gemini_enterprise_app_name: str | None = None,
     express_mode: bool = False,
+    avatar_config: types.AvatarConfig | None = None,
 ):
   """Starts a FastAPI server for agents.
 
@@ -2223,6 +2268,7 @@ async def _lifespan(app: FastAPI) -> AsyncIterator[None]:
           trigger_sources=trigger_sources,
           gemini_enterprise_app_name=gemini_enterprise_app_name,
           express_mode=express_mode,
+          avatar_config=avatar_config,
           lifespan=_lifespan,
       ),
       host=host,

src/google/adk/cli/fast_api.py

@@ -35,6 +35,7 @@
 from fastapi.encoders import jsonable_encoder
 from fastapi.responses import JSONResponse
 from fastapi.responses import StreamingResponse
+from google.genai import types
 from opentelemetry import context
 from opentelemetry import trace
 from opentelemetry.sdk.trace import export
@@ -122,6 +123,7 @@ def get_fast_api_app(
     default_llm_model: str | None = None,
     gemini_enterprise_app_name: str | None = None,
     express_mode: bool = False,
+    avatar_config: types.AvatarConfig | None = None,
 ) -> FastAPI:
   """Constructs and returns a FastAPI application for serving ADK agents.
 
@@ -178,6 +180,7 @@ def get_fast_api_app(
     gemini_enterprise_app_name: The Gemini Enterprise app name to use for the
       agent.
     express_mode: Whether to enable express mode.
+    avatar_config: Avatar configuration to apply to live agent runs.
 
   Returns:
     The configured FastAPI application instance.
@@ -300,6 +303,7 @@ def get_fast_api_app(
       auto_create_session=auto_create_session,
       trigger_sources=trigger_sources,
       default_llm_model=default_llm_model,
+      avatar_config=avatar_config,
   )
 
   # In single agent mode, use that agent as the default app.

tests/unittests/cli/test_adk_web_server_run_live.py

@@ -21,6 +21,7 @@
 from google.adk.cli.adk_web_server import AdkWebServer
 from google.adk.events.event import Event
 from google.adk.sessions.in_memory_session_service import InMemorySessionService
+from google.genai import types as genai_types
 import pytest
 from starlette.websockets import WebSocketDisconnect
 
@@ -61,7 +62,7 @@ async def run_live(
     yield Event(author="runner")
 
 
-def test_run_live_applies_run_config_query_options():
+def test_run_live_applies_server_and_query_run_config_options():
   session_service = InMemorySessionService()
   asyncio.run(
       session_service.create_session(
@@ -82,6 +83,7 @@ def test_run_live_applies_run_config_query_options():
       eval_sets_manager=types.SimpleNamespace(),
       eval_set_results_manager=types.SimpleNamespace(),
       agents_dir=".",
+      avatar_config=genai_types.AvatarConfig(avatar_name="Kai"),
   )
 
   async def _get_runner_async(_self, _app_name: str):
@@ -122,6 +124,8 @@ async def _get_runner_async(_self, _app_name: str):
   assert run_config.session_resumption.transparent is True
   assert run_config.save_live_blob is True
   assert run_config.explicit_vad_signal is True
+  assert run_config.avatar_config is not None
+  assert run_config.avatar_config.avatar_name == "Kai"
 
 
 @pytest.mark.parametrize(

tests/unittests/cli/utils/test_cli_tools_click.py

@@ -1589,6 +1589,28 @@ def test_cli_api_server_invokes_uvicorn(
   assert _patch_uvicorn.calls, "uvicorn.Server.run must be called"
 
 
+@pytest.mark.parametrize("command", ["web", "api_server"])
+def test_cli_server_passes_avatar_config(
+    tmp_path: Path,
+    _patch_uvicorn: _Recorder,
+    monkeypatch: pytest.MonkeyPatch,
+    command: str,
+) -> None:
+  """Both server commands pass parsed avatar configuration to the app."""
+  agents_dir = tmp_path / "agents"
+  agents_dir.mkdir()
+  mock_get_app = _Recorder()
+  monkeypatch.setattr("google.adk.cli.fast_api.get_fast_api_app", mock_get_app)
+
+  result = CliRunner().invoke(
+      cli_tools_click.main,
+      [command, "--avatar_config", '{"avatarName":"Kai"}', str(agents_dir)],
+  )
+
+  assert result.exit_code == 0, (result.output, repr(result.exception))
+  assert mock_get_app.calls[0][1]["avatar_config"].avatar_name == "Kai"
+
+
 def test_cli_web_passes_service_uris(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch, _patch_uvicorn: _Recorder
 ) -> None:
@@ -2632,10 +2654,41 @@ def test_fast_api_common_options_documented_defaults() -> None:
   assert captured["a2a"] is False
   assert captured["allow_origins"] == ()
   assert captured["log_level"] == "INFO"
+  assert captured["avatar_config"] is None
   # --verbose is consumed while folding it into log_level.
   assert "verbose" not in captured
 
 
+@pytest.mark.parametrize("from_file", [False, True])
+def test_fast_api_common_options_parses_avatar_config(
+    tmp_path: Path, from_file: bool
+) -> None:
+  """Avatar configuration accepts either inline JSON or a JSON file."""
+  command, captured = _fast_api_command()
+  config_json = '{"avatarName":"Kai","videoBitrateBps":1000000}'
+  value = config_json
+  if from_file:
+    config_path = tmp_path / "avatar.json"
+    config_path.write_text(config_json, encoding="utf-8")
+    value = str(config_path)
+
+  result = CliRunner().invoke(command, ["--avatar_config", value])
+
+  assert result.exit_code == 0, (result.output, repr(result.exception))
+  assert captured["avatar_config"].avatar_name == "Kai"
+  assert captured["avatar_config"].video_bitrate_bps == 1000000
+
+
+def test_fast_api_common_options_rejects_invalid_avatar_config() -> None:
+  """Invalid inline avatar JSON fails before either server starts."""
+  command, _ = _fast_api_command()
+
+  result = CliRunner().invoke(command, ["--avatar_config", "{invalid}"])
+
+  assert result.exit_code == 2
+  assert "valid AvatarConfig JSON object" in result.output
+
+
 # adk test
 @pytest.fixture
 def fake_pytest_run(monkeypatch: pytest.MonkeyPatch):