feat: Add telemetry consent configuration endpoints and local writing utility

- Establish read_telemetry_consent and write_telemetry_consent utilities to store opt-in status locally in ~/.adk/config.json.
- Implement GET and POST '/config/telemetry' FastAPI endpoints inside dev_server.py.
- Prevent CSRF/XSRF forgery by requiring the 'x-adk-telemetry-request: true' header on all POST requests.
- Add unit tests verifying route access permissions and json persistence.

Co-authored-by: Lucas Kang <lucaskang@google.com>
PiperOrigin-RevId: 952873455

Author: lkang172

src/google/adk/cli/api_server.py

@@ -80,6 +80,7 @@
 from ..runners import Runner
 from ..sessions.base_session_service import BaseSessionService
 from ..sessions.session import Session
+from ..utils._telemetry_config import read_telemetry_consent
 from ..utils.agent_info import AgentInfo
 from ..utils.agent_info import get_agents_dict
 from ..utils.context_utils import Aclosing
@@ -904,6 +905,9 @@ def _setup_runtime_config(self, web_assets_dir: str):
           runtime_config_path,
       )
     runtime_config["backendUrl"] = self.url_prefix if self.url_prefix else ""
+    # Inject telemetry consent on bootstrapping to avoid an extra API call
+    # when loading the UI.
+    runtime_config["telemetry"] = read_telemetry_consent()
 
     # Set custom logo config.
     if self.logo_text or self.logo_image_url:

src/google/adk/cli/browser/assets/config/runtime-config.json

@@ -1,3 +1,4 @@
 {
-  "backendUrl": ""
+  "backendUrl": "",
+  "telemetry": null
 }

src/google/adk/cli/dev_server.py

@@ -36,6 +36,7 @@
 
 from fastapi import FastAPI
 from fastapi import HTTPException
+from fastapi import Request as FastAPIRequest
 from fastapi import UploadFile
 from fastapi.responses import FileResponse
 from fastapi.responses import PlainTextResponse
@@ -59,6 +60,8 @@
 from ..evaluation.eval_metrics import MetricInfo
 from ..evaluation.eval_result import EvalSetResult
 from ..evaluation.eval_set import EvalSet
+from ..utils._telemetry_config import read_telemetry_consent
+from ..utils._telemetry_config import write_telemetry_consent
 from .api_server import ApiServer
 
 NESTED_APP_SEPARATOR = "."
@@ -152,6 +155,12 @@ class ListMetricsInfoResponse(common.BaseModel):
   metrics_info: list[MetricInfo]
 
 
+class TelemetryConsentRequest(common.BaseModel):
+  """Request body for setting the telemetry consent configuration."""
+
+  telemetry: bool
+
+
 class DevServer(ApiServer):
   """Development server that extends ApiServer with dev-only endpoints.
 
@@ -204,10 +213,28 @@ def _register_dev_endpoints(
   ):
     """Register all development-only endpoints.
 
-    This includes debug, evaluation, and graph visualization endpoints.
-    These endpoints should NOT be exposed in production deployments.
+    This includes debug, evaluation, graph visualization, and telemetry consent
+    endpoints. These endpoints should NOT be exposed in production deployments.
     """
 
+    @app.get("/config/telemetry")
+    async def get_telemetry_consent() -> dict[str, Any]:
+      """Gets the user configuration for telemetry consent."""
+      return {"telemetry": read_telemetry_consent()}
+
+    @app.post("/config/telemetry")
+    async def set_telemetry_consent(
+        req: TelemetryConsentRequest, request: FastAPIRequest
+    ) -> dict[str, Any]:
+      """Sets the user configuration for telemetry consent."""
+      if request.headers.get("x-adk-telemetry-request") != "true":
+        raise HTTPException(
+            status_code=400,
+            detail="Forbidden: missing required security header",
+        )
+      write_telemetry_consent(req.telemetry)
+      return {"telemetry": req.telemetry}
+
     # Import needed for eval endpoints
     from ..evaluation.constants import MISSING_EVAL_DEPENDENCIES_MESSAGE
 

src/google/adk/utils/_telemetry_config.py

@@ -0,0 +1,76 @@
+# 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.
+
+"""Telemetry user consent configuration utilities."""
+
+from __future__ import annotations
+
+import json
+import logging
+import pathlib
+import typing
+
+logger = logging.getLogger("google_adk." + __name__)
+
+
+def get_user_config_path() -> pathlib.Path:
+  """Returns the path to the ADK global config file."""
+  return pathlib.Path.home() / ".adk" / "config.json"
+
+
+def read_telemetry_consent() -> typing.Optional[bool]:
+  """Reads the telemetry consent status from local config (config.json).
+
+  Returns:
+    True if opted-in, False if opted-out, and None if no explicit
+    preference has been recorded yet or if there is an error reading
+    the config.
+  """
+  path = get_user_config_path()
+  if not path.exists():
+    return None
+  try:
+    with open(path, "r", encoding="utf-8") as f:
+      config = json.load(f)
+      val = config.get("telemetry", None)
+      if isinstance(val, bool):
+        return val
+      return None
+  except Exception as e:
+    logger.warning("Failed to read telemetry config from %s: %s", path, e)
+    return None
+
+
+def write_telemetry_consent(enabled: bool) -> None:
+  """Writes the telemetry consent status to local config (config.json)."""
+  path = get_user_config_path()
+  try:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    config = {}
+    if path.exists():
+      try:
+        with open(path, "r", encoding="utf-8") as f:
+          config = json.load(f)
+          if not isinstance(config, dict):
+            config = {}
+      except Exception:
+        # If config parsing fails, start with an empty dictionary
+        config = {}
+    config["telemetry"] = enabled
+    with open(path, "w", encoding="utf-8") as f:
+      json.dump(config, f, indent=2)
+      f.write("\n")
+  except Exception as e:
+    logger.error("Failed to write telemetry config to %s: %s", path, e)
+    raise

tests/unittests/cli/test_fast_api.py

@@ -617,8 +617,8 @@ def test_agent_with_bigquery_analytics_plugin(
           os.path,
           "exists",
           autospec=True,
-          side_effect=lambda p: p.endswith("plugins.yaml")
-          or p.endswith("root_agent.yaml"),
+          side_effect=lambda p: str(p).endswith("plugins.yaml")
+          or str(p).endswith("root_agent.yaml"),
       ),
   ):
     from google.adk.cli.adk_web_server import AdkWebServer
@@ -2862,6 +2862,35 @@ def test_version_endpoint(test_app):
   assert "language_version" in data
 
 
+def test_telemetry_get_endpoint(test_app):
+  """Test the GET telemetry consent endpoint."""
+  with patch(
+      "google.adk.cli.dev_server.read_telemetry_consent", return_value=True
+  ):
+    response = test_app.get("/config/telemetry")
+    assert response.status_code == 200
+    assert response.json() == {"telemetry": True}
+
+
+def test_telemetry_post_endpoint_success(test_app):
+  """Test the POST telemetry consent endpoint with required header."""
+  with patch("google.adk.cli.dev_server.write_telemetry_consent") as mock_write:
+    headers = {"x-adk-telemetry-request": "true"}
+    response = test_app.post(
+        "/config/telemetry", json={"telemetry": True}, headers=headers
+    )
+    assert response.status_code == 200
+    assert response.json() == {"telemetry": True}
+    mock_write.assert_called_once_with(True)
+
+
+def test_telemetry_post_endpoint_missing_header(test_app):
+  """Test the POST telemetry consent endpoint without required header."""
+  response = test_app.post("/config/telemetry", json={"telemetry": True})
+  assert response.status_code == 400
+  assert "Forbidden: missing required security header" in response.text
+
+
 @pytest.fixture
 def test_app_auto_session(
     mock_session_service,
@@ -3028,8 +3057,8 @@ async def run_async_capture(
           os.path,
           "exists",
           autospec=True,
-          side_effect=lambda p: "yaml_app" in p
-          and p.endswith("root_agent.yaml"),
+          side_effect=lambda p: "yaml_app" in str(p)
+          and str(p).endswith("root_agent.yaml"),
       ),
   ):
     app = get_fast_api_app(

tests/unittests/utils/test_telemetry_config.py

@@ -0,0 +1,102 @@
+# 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.
+
+"""Tests for telemetry_config.py."""
+
+import json
+import pathlib
+import unittest.mock
+
+from google.adk.utils import _telemetry_config as telemetry_config
+import pytest
+
+
+def test_get_user_config_path():
+  path = telemetry_config.get_user_config_path()
+  assert isinstance(path, pathlib.Path)
+  assert path.name == "config.json"
+  assert path.parent.name == ".adk"
+
+
+def test_read_telemetry_consent_not_exists(tmp_path):
+  config_file = tmp_path / "config.json"
+  with unittest.mock.patch.object(
+      telemetry_config, "get_user_config_path", return_value=config_file
+  ):
+    assert telemetry_config.read_telemetry_consent() is None
+
+
+def test_read_telemetry_consent_exists_true(tmp_path):
+  config_file = tmp_path / "config.json"
+  config_file.parent.mkdir(parents=True, exist_ok=True)
+  with open(config_file, "w", encoding="utf-8") as f:
+    json.dump({"telemetry": True}, f)
+
+  with unittest.mock.patch.object(
+      telemetry_config, "get_user_config_path", return_value=config_file
+  ):
+    assert telemetry_config.read_telemetry_consent() is True
+
+
+def test_read_telemetry_consent_exists_false(tmp_path):
+  config_file = tmp_path / "config.json"
+  config_file.parent.mkdir(parents=True, exist_ok=True)
+  with open(config_file, "w", encoding="utf-8") as f:
+    json.dump({"telemetry": False}, f)
+
+  with unittest.mock.patch.object(
+      telemetry_config, "get_user_config_path", return_value=config_file
+  ):
+    assert telemetry_config.read_telemetry_consent() is False
+
+
+def test_read_telemetry_consent_invalid_json(tmp_path):
+  config_file = tmp_path / "config.json"
+  config_file.parent.mkdir(parents=True, exist_ok=True)
+  with open(config_file, "w", encoding="utf-8") as f:
+    f.write("not raw json data")
+
+  with unittest.mock.patch.object(
+      telemetry_config, "get_user_config_path", return_value=config_file
+  ):
+    assert telemetry_config.read_telemetry_consent() is None
+
+
+def test_write_telemetry_consent(tmp_path):
+  config_file = tmp_path / "config.json"
+  with unittest.mock.patch.object(
+      telemetry_config, "get_user_config_path", return_value=config_file
+  ):
+    telemetry_config.write_telemetry_consent(True)
+    assert telemetry_config.read_telemetry_consent() is True
+
+    # Overwrite
+    telemetry_config.write_telemetry_consent(False)
+    assert telemetry_config.read_telemetry_consent() is False
+
+    with open(config_file, "r", encoding="utf-8") as f:
+      data = json.load(f)
+      assert data == {"telemetry": False}
+
+
+def test_write_telemetry_consent_raises_on_error(tmp_path):
+  config_file = tmp_path / "config.json"
+  with unittest.mock.patch.object(
+      telemetry_config, "get_user_config_path", return_value=config_file
+  ):
+    with unittest.mock.patch.object(
+        pathlib.Path, "mkdir", side_effect=OSError("Permission denied")
+    ):
+      with pytest.raises(OSError):
+        telemetry_config.write_telemetry_consent(True)