feat: Wire environment service into genai sdk

google/genai/client.py

@@ -41,10 +41,12 @@
 
 from ._gaos.google_genai import (
     AsyncGeminiNextGenAgents,
+    AsyncGeminiNextGenEnvironments,
     AsyncGeminiNextGenInteractions,
     AsyncGeminiNextGenTriggers,
     AsyncGeminiNextGenWebhooks,
     GeminiNextGenAgents,
+    GeminiNextGenEnvironments,
     GeminiNextGenInteractions,
     GeminiNextGenTriggers,
     GeminiNextGenWebhooks,
@@ -56,6 +58,7 @@
 
 _agent_experimental_warned = False
 _trigger_experimental_warned = False
+_environment_experimental_warned = False
 
 
 class AsyncClient:
@@ -78,6 +81,7 @@ def __init__(self, api_client: BaseApiClient):
     self._interactions: Optional[AsyncGeminiNextGenInteractions] = None
     self._webhooks: Optional[AsyncGeminiNextGenWebhooks] = None
     self._triggers: Optional[AsyncGeminiNextGenTriggers] = None
+    self._environments: Optional[AsyncGeminiNextGenEnvironments] = None
 
   @property
   def _nextgen_client(self) -> AsyncGeminiNextGenAPI:
@@ -127,6 +131,22 @@ def triggers(self) -> AsyncGeminiNextGenTriggers:
       self._triggers = AsyncGeminiNextGenTriggers(self._api_client)
     return self._triggers
 
+  @property
+  def environments(self) -> AsyncGeminiNextGenEnvironments:
+    """Environments resource."""
+    global _environment_experimental_warned
+    if not _environment_experimental_warned:
+      _environment_experimental_warned = True
+      warnings.warn(
+          'Environments usage is experimental and may change in future versions.',
+          category=UserWarning,
+          stacklevel=1,
+      )
+    if self._environments is None:
+      self._environments = AsyncGeminiNextGenEnvironments(self._api_client)
+    return self._environments
+
+
   @property
   def models(self) -> AsyncModels:
     return self._models
@@ -380,6 +400,7 @@ def __init__(
     self._interactions: Optional[GeminiNextGenInteractions] = None
     self._webhooks: Optional[GeminiNextGenWebhooks] = None
     self._triggers: Optional[GeminiNextGenTriggers] = None
+    self._environments: Optional[GeminiNextGenEnvironments] = None
 
   @staticmethod
   def _get_api_client(
@@ -465,6 +486,20 @@ def triggers(self) -> GeminiNextGenTriggers:
       self._triggers = GeminiNextGenTriggers(self._api_client)
     return self._triggers
 
+  @property
+  def environments(self) -> GeminiNextGenEnvironments:
+    global _environment_experimental_warned
+    if not _environment_experimental_warned:
+      _environment_experimental_warned = True
+      warnings.warn(
+          'Environments usage is experimental and may change in future versions.',
+          category=UserWarning,
+          stacklevel=2,
+      )
+    if self._environments is None:
+      self._environments = GeminiNextGenEnvironments(self._api_client)
+    return self._environments
+
   @property
   def chats(self) -> Chats:
     return Chats(modules=self.models)

google/genai/interactions.py

@@ -26,6 +26,8 @@
 # the name collision.
 from ._gaos.types.triggers import *  # noqa: F401,F403
 from ._gaos.types.triggers import __all__ as _triggers_all
+from ._gaos.types.environments import *  # noqa: F401,F403
+from ._gaos.types.environments import __all__ as _environments_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
@@ -137,4 +139,4 @@ class InteractionGetParamsStreaming(InteractionGetParamsBase):
 ]
 # 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)
+__all__ = __all__ + list(_triggers_all) + list(_resources_all) + list(_environments_all) + list(_interactions_all)

google/genai/tests/gaos/test_environments_lifecycle.py

@@ -0,0 +1,118 @@
+# 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 Environments API."""
+
+from __future__ import annotations
+
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+import json
+import threading
+
+import pytest
+
+from ... import Client
+
+ENVIRONMENT_BODY = {
+    "id": "env_abc_1234",
+    "status": "active",
+    "created": "2026-07-22T15:18:38Z",
+    "updated": "2026-07-22T15:18:38Z",
+    "sources": [
+        {
+            "type": "INLINE",
+            "content": "print('hello')",
+            "target": "main.py",
+        }
+    ],
+}
+
+
+class _RecordingHandler(BaseHTTPRequestHandler):
+  captured: list[str] = []
+  captured_bodies: list[dict] = []
+
+  def _record_and_respond(self) -> None:
+    self.captured.append(f"{self.command} {self.path}")
+    if self.command in ("POST", "PATCH", "PUT"):
+      content_length = int(self.headers.get("Content-Length", 0))
+      if content_length > 0:
+        body = self.rfile.read(content_length)
+        self.captured_bodies.append(json.loads(body.decode("utf-8")))
+    payload = json.dumps(ENVIRONMENT_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_environments_lifecycle_routes_through_google_genai_client(
+    monkeypatch,
+):
+  monkeypatch.delenv("GOOGLE_GENAI_USE_VERTEXAI", raising=False)
+  captured: list[str] = []
+  captured_bodies: list[dict] = []
+  handler = type("Handler", (_RecordingHandler,), {
+      "captured": captured,
+      "captured_bodies": captured_bodies,
+  })
+  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}",
+        },
+    )
+
+    environment = client.environments.create(
+        sources=[
+            {
+                "type": "inline",
+                "content": "print('hello')",
+                "target": "main.py",
+            }
+        ]
+    )
+    client.environments.list()
+    fetched = client.environments.get(id="env_abc_1234")
+    client.environments.delete(id="env_abc_1234")
+
+    assert environment.id == "env_abc_1234"
+    assert fetched.id == "env_abc_1234"
+    assert captured == [
+        "POST /v1beta/environments",
+        "GET /v1beta/environments",
+        "GET /v1beta/environments/env_abc_1234",
+        "DELETE /v1beta/environments/env_abc_1234",
+    ]
+
+    create_body = captured_bodies[0]
+    assert create_body["sources"][0]["content"] == "print('hello')"
+
+  finally:
+    server.shutdown()
+    thread.join()
+    server.server_close()