feat: add ADK trigger endpoints to ADK Web Server

PiperOrigin-RevId: 872815602

Author: google-genai-bot

src/google/adk/cli/adk_web_server.py

@@ -651,6 +651,7 @@ def __init__(
       logo_image_url: Optional[str] = None,
       url_prefix: Optional[str] = None,
       auto_create_session: bool = False,
+      trigger_sources: Optional[list[str]] = None,
   ):
     self.agent_loader = agent_loader
     self.session_service = session_service
@@ -669,6 +670,7 @@ def __init__(
     self.runner_dict = {}
     self.url_prefix = url_prefix
     self.auto_create_session = auto_create_session
+    self.trigger_sources = trigger_sources
 
   async def get_runner_async(self, app_name: str) -> Runner:
     """Returns the cached runner for the given app."""
@@ -2058,6 +2060,12 @@ async def process_messages():
         for task in pending:
           task.cancel()
 
+    # Register /trigger/* endpoints.
+    from .trigger_routes import TriggerRouter
+
+    trigger_router = TriggerRouter(self, trigger_sources=self.trigger_sources)
+    trigger_router.register(app)
+
     if web_assets_dir:
       import mimetypes
 

src/google/adk/cli/cli_deploy.py

@@ -99,7 +99,7 @@ def _ensure_agent_engine_dependency(requirements_txt_path: str) -> None:
 
 EXPOSE {port}
 
-CMD adk {command} --port={port} {host_option} {service_option} {trace_to_cloud_option} {otel_to_cloud_option} {allow_origins_option} {a2a_option} "/app/agents"
+CMD {cmd_exec}
 """
 
 _AGENT_ENGINE_APP_TEMPLATE: Final[str] = """
@@ -591,8 +591,8 @@ def _get_service_option_by_adk_version(
     artifact_uri: Optional[str],
     memory_uri: Optional[str],
     use_local_storage: Optional[bool] = None,
-) -> str:
-  """Returns service option string based on adk_version."""
+) -> list[str]:
+  """Returns service options based on adk_version."""
   parsed_version = parse(adk_version)
   options: list[str] = []
 
@@ -621,7 +621,7 @@ def _get_service_option_by_adk_version(
           else '--no_use_local_storage'
       ))
 
-  return ' '.join(options)
+  return options
 
 
 def to_cloud_run(
@@ -645,6 +645,7 @@ def to_cloud_run(
     memory_service_uri: Optional[str] = None,
     use_local_storage: bool = False,
     a2a: bool = False,
+    trigger_sources: Optional[str] = None,
     extra_gcloud_args: Optional[tuple[str, ...]] = None,
 ):
   """Deploys an agent to Google Cloud Run.
@@ -710,31 +711,45 @@ def to_cloud_run(
 
     # create Dockerfile
     click.echo('Creating Dockerfile...')
-    host_option = '--host=0.0.0.0' if adk_version > '0.5.0' else ''
-    allow_origins_option = (
-        f'--allow_origins={",".join(allow_origins)}' if allow_origins else ''
+    cmd_args = [
+        'adk',
+        'web' if with_ui else 'api_server',
+        f'--port={port}',
+    ]
+    if adk_version > '0.5.0':
+      cmd_args.append('--host=0.0.0.0')
+
+    cmd_args.extend(
+        _get_service_option_by_adk_version(
+            adk_version,
+            session_service_uri,
+            artifact_service_uri,
+            memory_service_uri,
+            use_local_storage,
+        )
     )
-    a2a_option = '--a2a' if a2a else ''
+
+    if trace_to_cloud:
+      cmd_args.append('--trace_to_cloud')
+    if otel_to_cloud:
+      cmd_args.append('--otel_to_cloud')
+    if allow_origins:
+      cmd_args.append(f'--allow_origins={",".join(allow_origins)}')
+    if a2a:
+      cmd_args.append('--a2a')
+    if trigger_sources:
+      cmd_args.append(f'--trigger_sources={trigger_sources}')
+
+    cmd_args.append('/app/agents')
+
     dockerfile_content = _DOCKERFILE_TEMPLATE.format(
         gcp_project_id=project,
         gcp_region=region,
         app_name=app_name,
         port=port,
-        command='web' if with_ui else 'api_server',
         install_agent_deps=install_agent_deps,
-        service_option=_get_service_option_by_adk_version(
-            adk_version,
-            session_service_uri,
-            artifact_service_uri,
-            memory_service_uri,
-            use_local_storage,
-        ),
-        trace_to_cloud_option='--trace_to_cloud' if trace_to_cloud else '',
-        otel_to_cloud_option='--otel_to_cloud' if otel_to_cloud else '',
-        allow_origins_option=allow_origins_option,
         adk_version=adk_version,
-        host_option=host_option,
-        a2a_option=a2a_option,
+        cmd_exec=json.dumps(cmd_args),
     )
     dockerfile_path = os.path.join(temp_folder, 'Dockerfile')
     os.makedirs(temp_folder, exist_ok=True)
@@ -1171,6 +1186,7 @@ def to_gke(
     memory_service_uri: Optional[str] = None,
     use_local_storage: bool = False,
     a2a: bool = False,
+    trigger_sources: Optional[str] = None,
     service_type: Literal[
         'ClusterIP', 'NodePort', 'LoadBalancer'
     ] = 'ClusterIP',
@@ -1247,27 +1263,45 @@ def to_gke(
     # create Dockerfile
     click.secho('\nSTEP 2: Generating deployment files...', bold=True)
     click.echo('  - Creating Dockerfile...')
-    host_option = '--host=0.0.0.0' if adk_version > '0.5.0' else ''
+    cmd_args = [
+        'adk',
+        'web' if with_ui else 'api_server',
+        f'--port={port}',
+    ]
+    if adk_version > '0.5.0':
+      cmd_args.append('--host=0.0.0.0')
+
+    cmd_args.extend(
+        _get_service_option_by_adk_version(
+            adk_version,
+            session_service_uri,
+            artifact_service_uri,
+            memory_service_uri,
+            use_local_storage,
+        )
+    )
+
+    if trace_to_cloud:
+      cmd_args.append('--trace_to_cloud')
+    if otel_to_cloud:
+      cmd_args.append('--otel_to_cloud')
+    if allow_origins:
+      cmd_args.append(f'--allow_origins={",".join(allow_origins)}')
+    if a2a:
+      cmd_args.append('--a2a')
+    if trigger_sources:
+      cmd_args.append(f'--trigger_sources={trigger_sources}')
+
+    cmd_args.append('/app/agents')
+
     dockerfile_content = _DOCKERFILE_TEMPLATE.format(
         gcp_project_id=project,
         gcp_region=region,
         app_name=app_name,
         port=port,
-        command='web' if with_ui else 'api_server',
         install_agent_deps=install_agent_deps,
-        service_option=_get_service_option_by_adk_version(
-            adk_version,
-            session_service_uri,
-            artifact_service_uri,
-            memory_service_uri,
-            use_local_storage,
-        ),
-        trace_to_cloud_option='--trace_to_cloud' if trace_to_cloud else '',
-        otel_to_cloud_option='--otel_to_cloud' if otel_to_cloud else '',
-        allow_origins_option=allow_origins_option,
         adk_version=adk_version,
-        host_option=host_option,
-        a2a_option='--a2a' if a2a else '',
+        cmd_exec=json.dumps(cmd_args),
     )
     dockerfile_path = os.path.join(temp_folder, 'Dockerfile')
     os.makedirs(temp_folder, exist_ok=True)

src/google/adk/cli/cli_tools_click.py

@@ -1329,6 +1329,7 @@ def fast_api_common_options():
   """Decorator to add common fast api options to click commands."""
 
   def decorator(func):
+
     @click.option(
         "--host",
         type=str,
@@ -1433,6 +1434,17 @@ def decorator(func):
         ),
         default=None,
     )
+    # Parsed into list[str] by the wrapper below (server commands need a list).
+    @click.option(
+        "--trigger_sources",
+        type=str,
+        help=(
+            "Optional. Comma-separated list of trigger sources to enable"
+            " (e.g., 'pubsub,eventarc'). Registers /apps/{app_name}/trigger/*"
+            " endpoints for batch and event-driven agent invocations."
+        ),
+        default=None,
+    )
     @functools.wraps(func)
     @click.pass_context
     def wrapper(ctx, *args, **kwargs):
@@ -1444,6 +1456,21 @@ def wrapper(ctx, *args, **kwargs):
       ):
         kwargs["log_level"] = "DEBUG"
 
+      # Parse comma-separated trigger_sources into a list and validate.
+      trigger_sources_raw = kwargs.get("trigger_sources")
+      if trigger_sources_raw is not None:
+        valid_sources = ["pubsub", "eventarc"]
+        parsed_sources = [
+            s.strip() for s in trigger_sources_raw.split(",") if s.strip()
+        ]
+        invalid = [s for s in parsed_sources if s not in valid_sources]
+        if invalid:
+          raise click.BadParameter(
+              f"Invalid trigger source(s): {', '.join(invalid)}. "
+              f"Valid sources are: {', '.join(valid_sources)}"
+          )
+        kwargs["trigger_sources"] = parsed_sources
+
       return func(*args, **kwargs)
 
     return wrapper
@@ -1486,6 +1513,7 @@ def cli_web(
     extra_plugins: Optional[list[str]] = None,
     logo_text: Optional[str] = None,
     logo_image_url: Optional[str] = None,
+    trigger_sources: Optional[list[str]] = None,
 ):
   """Starts a FastAPI server with Web UI for agents.
 
@@ -1542,6 +1570,7 @@ async def _lifespan(app: FastAPI):
       extra_plugins=extra_plugins,
       logo_text=logo_text,
       logo_image_url=logo_image_url,
+      trigger_sources=trigger_sources,
   )
   config = uvicorn.Config(
       app,
@@ -1597,6 +1626,7 @@ def cli_api_server(
     reload_agents: bool = False,
     extra_plugins: Optional[list[str]] = None,
     auto_create_session: bool = False,
+    trigger_sources: Optional[list[str]] = None,
 ):
   """Starts a FastAPI server for agents.
 
@@ -1630,6 +1660,7 @@ def cli_api_server(
           reload_agents=reload_agents,
           extra_plugins=extra_plugins,
           auto_create_session=auto_create_session,
+          trigger_sources=trigger_sources,
       ),
       host=host,
       port=port,
@@ -1763,6 +1794,17 @@ def cli_api_server(
     default=False,
     help="Optional. Whether to enable A2A endpoint.",
 )
+# Kept as raw str (not parsed to list) — interpolated directly into Dockerfile CMD.
+@click.option(
+    "--trigger_sources",
+    type=str,
+    help=(
+        "Optional. Comma-separated list of trigger sources to enable"
+        " (e.g., 'pubsub,eventarc'). Registers /trigger/* endpoints"
+        " for batch and event-driven agent invocations."
+    ),
+    default=None,
+)
 @click.option(
     "--allow_origins",
     help=(
@@ -1799,6 +1841,7 @@ def cli_deploy_cloud_run(
     session_db_url: Optional[str] = None,  # Deprecated
     artifact_storage_uri: Optional[str] = None,  # Deprecated
     a2a: bool = False,
+    trigger_sources: Optional[str] = None,
 ):
   """Deploys an agent to Cloud Run.
 
@@ -1876,6 +1919,7 @@ def cli_deploy_cloud_run(
         memory_service_uri=memory_service_uri,
         use_local_storage=use_local_storage,
         a2a=a2a,
+        trigger_sources=trigger_sources,
         extra_gcloud_args=tuple(gcloud_args),
     )
   except Exception as e:
@@ -2272,6 +2316,17 @@ def cli_deploy_agent_engine(
         " version in the dev environment)"
     ),
 )
+# Kept as raw str (not parsed to list) — interpolated directly into Dockerfile CMD.
+@click.option(
+    "--trigger_sources",
+    type=str,
+    help=(
+        "Optional. Comma-separated list of trigger sources to enable"
+        " (e.g., 'pubsub,eventarc'). Registers /trigger/* endpoints"
+        " for batch and event-driven agent invocations."
+    ),
+    default=None,
+)
 @adk_services_options(default_use_local_storage=False)
 @click.argument(
     "agent",
@@ -2298,6 +2353,7 @@ def cli_deploy_gke(
     artifact_service_uri: Optional[str] = None,
     memory_service_uri: Optional[str] = None,
     use_local_storage: bool = False,
+    trigger_sources: Optional[str] = None,
 ):
   """Deploys an agent to GKE.
 
@@ -2329,6 +2385,7 @@ def cli_deploy_gke(
         artifact_service_uri=artifact_service_uri,
         memory_service_uri=memory_service_uri,
         use_local_storage=use_local_storage,
+        trigger_sources=trigger_sources,
     )
   except Exception as e:
     click.secho(f"Deploy failed: {e}", fg="red", err=True)

src/google/adk/cli/fast_api.py

@@ -94,6 +94,7 @@ def get_fast_api_app(
     logo_text: Optional[str] = None,
     logo_image_url: Optional[str] = None,
     auto_create_session: bool = False,
+    trigger_sources: Optional[list[str]] = None,
 ) -> FastAPI:
   """Constructs and returns a FastAPI application for serving ADK agents.
 
@@ -135,8 +136,11 @@ def get_fast_api_app(
     extra_plugins: List of extra plugin names to load.
     logo_text: Text to display in the web UI logo area.
     logo_image_url: URL for an image to display in the web UI logo area.
-    auto_create_session: Whether to automatically create a session when
-      not found.
+    auto_create_session: Whether to automatically create a session when not
+      found.
+    trigger_sources: List of trigger sources to enable (e.g. ["pubsub",
+      "eventarc"]). When set, registers /trigger/* endpoints for batch and
+      event-driven agent invocations. None disables all trigger endpoints.
 
   Returns:
     The configured FastAPI application instance.
@@ -205,6 +209,7 @@ def get_fast_api_app(
       logo_image_url=logo_image_url,
       url_prefix=url_prefix,
       auto_create_session=auto_create_session,
+      trigger_sources=trigger_sources,
   )
 
   # Callbacks & other optional args for when constructing the FastAPI instance