feat: Add Triggers resource to GoogleGenAI client.

google/genai/_gaos/google_genai.py

@@ -48,6 +48,8 @@
 from .types import interactions
 from .types.security import Security
 from .utils import BackoffStrategy, RetryConfig, eventstreaming
+from .triggers import AsyncTriggers as GeneratedAsyncTriggers
+from .triggers import Triggers as GeneratedTriggers
 from .webhooks import AsyncWebhooks as GeneratedAsyncWebhooks
 from .webhooks import Webhooks as GeneratedWebhooks
 
@@ -667,6 +669,88 @@ async def delete(self, *args: Any, **kwargs: Any) -> Any:
             return await async_wrap_sdk_call(super().delete, *args, **kwargs)
 
 
+class GeminiNextGenTriggers(GeneratedTriggers):
+    """Public triggers resource backed by the NextGen client.
+
+    Subclasses the generated resource so every public method is wrapped in
+    `wrap_sdk_call`, translating per-operation `GenAiError` raises into the
+    status-code `APIError` hierarchy exposed at the
+    `google.genai._interactions` import surface.
+    """
+
+    def __init__(self, api_client: Any):
+        sdk = build_google_genai_client(api_client)
+        super().__init__(sdk.sdk_configuration, parent_ref=sdk)
+
+    if not TYPE_CHECKING:
+        @property
+        def with_raw_response(self):
+            return _RawResponseAccessorProxy(super().with_raw_response)
+
+        @property
+        def with_streaming_response(self):
+            return _RawResponseAccessorProxy(super().with_streaming_response)
+
+        def create(self, *args: Any, **kwargs: Any) -> Any:
+            return wrap_sdk_call(super().create, *args, **kwargs)
+
+        def list(self, *args: Any, **kwargs: Any) -> Any:
+            return wrap_sdk_call(super().list, *args, **kwargs)
+
+        def get(self, *args: Any, **kwargs: Any) -> Any:
+            return wrap_sdk_call(super().get, *args, **kwargs)
+
+        def update(self, *args: Any, **kwargs: Any) -> Any:
+            return wrap_sdk_call(super().update, *args, **kwargs)
+
+        def delete(self, *args: Any, **kwargs: Any) -> Any:
+            return wrap_sdk_call(super().delete, *args, **kwargs)
+
+        def run(self, *args: Any, **kwargs: Any) -> Any:
+            return wrap_sdk_call(super().run, *args, **kwargs)
+
+        def list_executions(self, *args: Any, **kwargs: Any) -> Any:
+            return wrap_sdk_call(super().list_executions, *args, **kwargs)
+
+
+class AsyncGeminiNextGenTriggers(GeneratedAsyncTriggers):
+    """Async public triggers resource backed by the NextGen client."""
+
+    def __init__(self, api_client: Any):
+        sdk = build_google_genai_async_client(api_client)
+        super().__init__(sdk.sdk_configuration, parent_ref=sdk)
+
+    if not TYPE_CHECKING:
+        @property
+        def with_raw_response(self):
+            return _AsyncRawResponseAccessorProxy(super().with_raw_response)
+
+        @property
+        def with_streaming_response(self):
+            return _AsyncRawResponseAccessorProxy(super().with_streaming_response)
+
+        async def create(self, *args: Any, **kwargs: Any) -> Any:
+            return await async_wrap_sdk_call(super().create, *args, **kwargs)
+
+        async def list(self, *args: Any, **kwargs: Any) -> Any:
+            return await async_wrap_sdk_call(super().list, *args, **kwargs)
+
+        async def get(self, *args: Any, **kwargs: Any) -> Any:
+            return await async_wrap_sdk_call(super().get, *args, **kwargs)
+
+        async def update(self, *args: Any, **kwargs: Any) -> Any:
+            return await async_wrap_sdk_call(super().update, *args, **kwargs)
+
+        async def delete(self, *args: Any, **kwargs: Any) -> Any:
+            return await async_wrap_sdk_call(super().delete, *args, **kwargs)
+
+        async def run(self, *args: Any, **kwargs: Any) -> Any:
+            return await async_wrap_sdk_call(super().run, *args, **kwargs)
+
+        async def list_executions(self, *args: Any, **kwargs: Any) -> Any:
+            return await async_wrap_sdk_call(super().list_executions, *args, **kwargs)
+
+
 def _add_output_properties_if_interaction(value: Any) -> Any:
     normalized = _normalize_interaction_shape(value)
     if normalized is None:

google/genai/client.py

@@ -42,9 +42,11 @@
 from ._gaos.google_genai import (
     AsyncGeminiNextGenAgents,
     AsyncGeminiNextGenInteractions,
+    AsyncGeminiNextGenTriggers,
     AsyncGeminiNextGenWebhooks,
     GeminiNextGenAgents,
     GeminiNextGenInteractions,
+    GeminiNextGenTriggers,
     GeminiNextGenWebhooks,
     build_google_genai_async_client,
     build_google_genai_client,
@@ -53,6 +55,7 @@
 from ._gaos.sdk import GenAI as GeminiNextGenAPI
 
 _agent_experimental_warned = False
+_trigger_experimental_warned = False
 
 
 class AsyncClient:
@@ -74,6 +77,7 @@ def __init__(self, api_client: BaseApiClient):
     self._agents: Optional[AsyncGeminiNextGenAgents] = None
     self._interactions: Optional[AsyncGeminiNextGenInteractions] = None
     self._webhooks: Optional[AsyncGeminiNextGenWebhooks] = None
+    self._triggers: Optional[AsyncGeminiNextGenTriggers] = None
 
   @property
   def _nextgen_client(self) -> AsyncGeminiNextGenAPI:
@@ -109,6 +113,20 @@ def agents(self) -> AsyncGeminiNextGenAgents:
       self._agents = AsyncGeminiNextGenAgents(self._api_client)
     return self._agents
 
+  @property
+  def triggers(self) -> AsyncGeminiNextGenTriggers:
+    global _trigger_experimental_warned
+    if not _trigger_experimental_warned:
+      _trigger_experimental_warned = True
+      warnings.warn(
+          'Triggers usage is experimental and may change in future versions.',
+          category=UserWarning,
+          stacklevel=1,
+      )
+    if self._triggers is None:
+      self._triggers = AsyncGeminiNextGenTriggers(self._api_client)
+    return self._triggers
+
   @property
   def models(self) -> AsyncModels:
     return self._models
@@ -361,6 +379,7 @@ def __init__(
     self._agents: Optional[GeminiNextGenAgents] = None
     self._interactions: Optional[GeminiNextGenInteractions] = None
     self._webhooks: Optional[GeminiNextGenWebhooks] = None
+    self._triggers: Optional[GeminiNextGenTriggers] = None
 
   @staticmethod
   def _get_api_client(
@@ -432,6 +451,20 @@ def agents(self) -> GeminiNextGenAgents:
       self._agents = GeminiNextGenAgents(self._api_client)
     return self._agents
 
+  @property
+  def triggers(self) -> GeminiNextGenTriggers:
+    global _trigger_experimental_warned
+    if not _trigger_experimental_warned:
+      _trigger_experimental_warned = True
+      warnings.warn(
+          'Triggers usage is experimental and may change in future versions.',
+          category=UserWarning,
+          stacklevel=2,
+      )
+    if self._triggers is None:
+      self._triggers = GeminiNextGenTriggers(self._api_client)
+    return self._triggers
+
   @property
   def chats(self) -> Chats:
     return Chats(modules=self.models)

google/genai/interactions.py

@@ -20,6 +20,12 @@
 
 from typing_extensions import Literal, Required, TypedDict
 
+# Import triggers before interactions so that interactions.Interaction (the
+# resource class) overrides triggers.Interaction (the TypeAliasType representing
+# nested interactions inside triggers) in the exported namespace, resolving
+# the name collision.
+from ._gaos.types.triggers import *  # noqa: F401,F403
+from ._gaos.types.triggers import __all__ as _triggers_all
 from ._gaos.types.interactions import *  # noqa: F401,F403
 from ._gaos.types.interactions import __all__ as _interactions_all
 from ._gaos.models.listagents import ListAgentsRequestParam as AgentListParams
@@ -129,4 +135,6 @@ class InteractionGetParamsStreaming(InteractionGetParamsBase):
     "WebhookRotateSigningSecretParams",
     "WebhookUpdateParams",
 ]
-__all__ = __all__ + list(_interactions_all) + list(_resources_all)
+# Ensure _interactions_all is appended last so interactions.Interaction wins
+# when doing wildcard imports from this module.
+__all__ = __all__ + list(_triggers_all) + list(_resources_all) + list(_interactions_all)

google/genai/tests/gaos/__init__.py

@@ -0,0 +1 @@
+# gaos tests package

google/genai/tests/gaos/test_triggers_lifecycle.py

@@ -0,0 +1,143 @@
+# Copyright 2026 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.
+"""Lifecycle tests for Triggers API."""
+
+from __future__ import annotations
+
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+import json
+import threading
+
+from ... import Client
+
+TRIGGER_BODY = {
+    "id": "projects/my-project/locations/my-location/triggers/svc_abc",
+    "schedule": "0 0 * * *",
+    "time_zone": "UTC",
+    "interaction": {
+        "agent": "projects/my-project/locations/my-location/agents/my-agent",
+        "input": "test-input",
+        "environment": {
+            "type": "remote",
+            "network": {
+                "allowlist": [
+                    {
+                        "domain": "api.github.com",
+                        "transform": {
+                            "Authorization": (
+                                "Bearer"
+                                " ghp_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
+                            )
+                        },
+                    },
+                    {"domain": "github.com"},
+                ]
+            },
+        },
+    },
+}
+
+
+class _RecordingHandler(BaseHTTPRequestHandler):
+  captured: list[str] = []
+
+  def _record_and_respond(self) -> None:
+    self.captured.append(f"{self.command} {self.path}")
+    payload = json.dumps(TRIGGER_BODY).encode()
+    self.send_response(200)
+    self.send_header("content-type", "application/json")
+    self.send_header("content-length", str(len(payload)))
+    self.end_headers()
+    self.wfile.write(payload)
+
+  do_GET = _record_and_respond
+  do_POST = _record_and_respond
+  do_PATCH = _record_and_respond
+  do_DELETE = _record_and_respond
+
+  def log_message(self, *args) -> None:
+    pass
+
+
+def test_python_triggers_lifecycle_routes_through_google_genai_client(
+    monkeypatch,
+):
+  monkeypatch.delenv("GOOGLE_GENAI_USE_VERTEXAI", raising=False)
+  captured: list[str] = []
+  handler = type("Handler", (_RecordingHandler,), {"captured": captured})
+  server = ThreadingHTTPServer(("127.0.0.1", 0), handler)
+  thread = threading.Thread(target=server.serve_forever, daemon=True)
+  thread.start()
+  try:
+    client = Client(
+        api_key="test-api-key",
+        http_options={
+            "api_version": "v1beta",
+            "base_url": f"http://127.0.0.1:{server.server_port}",
+        },
+    )
+
+    trigger = client.triggers.create(
+        interaction={
+            "agent": (
+                "projects/my-project/locations/my-location/agents/my-agent"
+            ),
+            "input": "test-input",
+            "environment": {
+                "type": "remote",
+                "network": {
+                    "allowlist": [
+                        {
+                            "domain": "api.github.com",
+                            "transform": {
+                                "Authorization": (
+                                    "Bearer"
+                                    " ghp_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
+                                )
+                            },
+                        },
+                        {"domain": "github.com"},
+                    ]
+                },
+            },
+        },
+        schedule="0 0 * * *",
+        time_zone="UTC",
+    )
+    client.triggers.list(filter_="some-filter", page_size=10)
+    fetched = client.triggers.get(id="svc_abc")
+    client.triggers.update(
+        id="svc_abc",
+        display_name="updated-name",
+        status="paused",
+    )
+    client.triggers.delete(id="svc_abc")
+    client.triggers.run(trigger_id="svc_abc")
+    client.triggers.list_executions(trigger_id="svc_abc", page_size=5)
+
+    assert trigger.schedule == "0 0 * * *"
+    assert fetched.schedule == "0 0 * * *"
+    assert captured == [
+        "POST /v1beta/triggers",
+        "GET /v1beta/triggers?filter=some-filter&page_size=10",
+        "GET /v1beta/triggers/svc_abc",
+        "PATCH /v1beta/triggers/svc_abc",
+        "DELETE /v1beta/triggers/svc_abc",
+        "POST /v1beta/triggers/svc_abc/executions",
+        "GET /v1beta/triggers/svc_abc/executions?page_size=5",
+    ]
+  finally:
+    server.shutdown()
+    thread.join()
+    server.server_close()