feat: add ADK trigger endpoints to ADK Web Server

src/google/adk/cli/adk_web_server.py

@@ -657,6 +657,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
@@ -675,6 +676,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."""
@@ -2114,6 +2116,13 @@ async def process_messages():
         for task in pending:
           task.cancel()
 
+    # Register /trigger/* endpoints when enabled.
+    if self.trigger_sources:
+      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 adk {command} --port={port} {host_option} {service_option} {trace_to_cloud_option} {otel_to_cloud_option} {allow_origins_option} {a2a_option} {trigger_sources_option} "/app/agents"
 """
 
 _AGENT_ENGINE_APP_TEMPLATE: Final[str] = """
@@ -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.
@@ -715,6 +716,9 @@ def to_cloud_run(
         f'--allow_origins={",".join(allow_origins)}' if allow_origins else ''
     )
     a2a_option = '--a2a' if a2a else ''
+    trigger_sources_option = (
+        f'--trigger_sources={trigger_sources}' if trigger_sources else ''
+    )
     dockerfile_content = _DOCKERFILE_TEMPLATE.format(
         gcp_project_id=project,
         gcp_region=region,
@@ -735,6 +739,7 @@ def to_cloud_run(
         adk_version=adk_version,
         host_option=host_option,
         a2a_option=a2a_option,
+        trigger_sources_option=trigger_sources_option,
     )
     dockerfile_path = os.path.join(temp_folder, 'Dockerfile')
     os.makedirs(temp_folder, exist_ok=True)
@@ -1178,6 +1183,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',
@@ -1275,6 +1281,9 @@ def to_gke(
         adk_version=adk_version,
         host_option=host_option,
         a2a_option='--a2a' if a2a else '',
+        trigger_sources_option=(
+            f'--trigger_sources={trigger_sources}' if trigger_sources else ''
+        ),
     )
     dockerfile_path = os.path.join(temp_folder, 'Dockerfile')
     os.makedirs(temp_folder, exist_ok=True)

src/google/adk/cli/cli_tools_click.py

@@ -1452,6 +1452,7 @@ def fast_api_common_options():
   """Decorator to add common fast api options to click commands."""
 
   def decorator(func):
+
     @click.option(
         "--host",
         type=str,
@@ -1556,6 +1557,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):
@@ -1567,6 +1579,13 @@ def wrapper(ctx, *args, **kwargs):
       ):
         kwargs["log_level"] = "DEBUG"
 
+      # Parse comma-separated trigger_sources into a list.
+      trigger_sources = kwargs.get("trigger_sources")
+      if trigger_sources is not None:
+        kwargs["trigger_sources"] = [
+            s.strip() for s in trigger_sources.split(",") if s.strip()
+        ]
+
       return func(*args, **kwargs)
 
     return wrapper
@@ -1609,6 +1628,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.
 
@@ -1665,6 +1685,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,
@@ -1720,6 +1741,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.
 
@@ -1753,6 +1775,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,
@@ -1887,6 +1910,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=(
@@ -1923,6 +1957,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.
 
@@ -2000,6 +2035,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:
@@ -2397,6 +2433,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",
@@ -2423,6 +2470,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.
 
@@ -2454,6 +2502,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

@@ -22,6 +22,7 @@
 import shutil
 import sys
 from typing import Any
+from typing import Literal
 from typing import Mapping
 from typing import Optional
 
@@ -95,6 +96,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[Literal["pubsub", "eventarc"]]] = None,
 ) -> FastAPI:
   """Constructs and returns a FastAPI application for serving ADK agents.
 
@@ -136,8 +138,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.
@@ -206,6 +211,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