feat: detect AI agent invocations in telemetry

madhavdonthula1 · madhavdonthula1 · commit 1e1f9c17c85d · 2026-07-14T17:08:48.000-07:00
Add an AgentDetector, modeled on the existing CICDDetector, that
identifies when SAM CLI is invoked by an AI coding agent by reading
environment variables the agent sets, and records it via the existing
userAgent telemetry field (emitted as "&lt;agent&gt;/1.0"). This lets us
distinguish agent-driven usage from human and CI/CD usage.

Detected agents and their signals:
  - Claude Code    CLAUDECODE present
  - Codex          any CODEX_* var present
  - Cursor         CURSOR_AGENT present
  - Gemini CLI     GEMINI_CLI present
  - Kiro           TERM_PROGRAM == kiro, or AWS_EXECUTION_ENV contains
                   amazonq/kiro
  - OpenCode       OPENCODE present
  - GitHub Copilot COPILOT_AGENT_SESSION_ID present, or
                   GITHUB_COPILOT_CLI_MODE == "true"

Shared variables (AWS_EXECUTION_ENV, TERM_PROGRAM) are substring or
exact-value matched, never presence-checked, to avoid misattributing
Lambda/CloudShell/CodeBuild or generic VS Code environments as agents.
GitHub Copilot detection deliberately avoids any GITHUB_* / GITHUB_ACTION
key so it never false-matches ordinary GitHub Actions CI, which the
existing CICDDetector already handles.

An existing AWS_TOOLING_USER_AGENT (AWS toolkit) takes precedence over a
detected agent. Amazon Q is intentionally deferred; until it is added
after Kiro in resolution order, Amazon Q CLI runs resolve to kiro.

Includes unit tests for each agent, the exact-name/too-generic guards
(GEMINI_CLI, OPENCODE vs AGENT), the AWS_EXECUTION_ENV and GITHUB_ACTION
false-positive guards, and toolkit-takes-precedence behavior.
diff --git a/samcli/lib/telemetry/agent_detector.py b/samcli/lib/telemetry/agent_detector.py
@@ -0,0 +1,104 @@
+"""
+Module used for detecting whether SAMCLI is executed by an AI agent.
+"""
+
+import os
+from enum import Enum, auto
+from typing import Callable, Dict, Mapping, Optional, Union
+
+
+class Agent(Enum):
+    ClaudeCode = auto()
+    Codex = auto()
+    Cursor = auto()
+    GeminiCLI = auto()
+    Kiro = auto()
+    OpenCode = auto()
+    GitHubCopilot = auto()
+
+
+def _is_codex(environ: Mapping) -> bool:
+    """
+    Whether it is running in Codex (OpenAI's Codex CLI). Codex sets a family of
+    CODEX_* variables rather than one canonical var (e.g. CODEX_SANDBOX, CODEX_CI,
+    CODEX_THREAD_ID), so match any name starting with "CODEX_".
+    """
+    return any(key.startswith("CODEX_") for key in environ)
+
+
+def _is_kiro(environ: Mapping) -> bool:
+    """
+    Whether it is running in Kiro (the Kiro IDE, via TERM_PROGRAM=kiro, or the Kiro
+    CLI, the renamed Amazon Q Developer CLI, via AWS_EXECUTION_ENV). AWS_EXECUTION_ENV
+    is shared with AWS Lambda/CloudShell/CodeBuild (e.g. "AWS_Lambda_python3.12"), so
+    it is substring-matched on "amazonq"/"kiro", not checked for presence, to avoid
+    misattributing those environments.
+
+    NOTE: the "amazonq" substring also matches the genuine Amazon Q Developer CLI.
+    Amazon Q is deferred for now, so its CLI runs are attributed to kiro until a
+    dedicated Amazon Q agent is added (which must be ordered AFTER Kiro in the enum).
+    """
+    if environ.get("TERM_PROGRAM", "") == "kiro":
+        return True
+    aws_execution_env: str = environ.get("AWS_EXECUTION_ENV", "").lower()
+    return "amazonq" in aws_execution_env or "kiro" in aws_execution_env
+
+
+_ENV_VAR_OR_CALLABLE_BY_AGENT: Dict[Agent, Union[str, Callable[[Mapping], bool]]] = {
+    # https://docs.anthropic.com/en/docs/claude-code
+    Agent.ClaudeCode: "CLAUDECODE",
+    # https://developers.openai.com/codex/cli
+    Agent.Codex: _is_codex,
+    # match presence, not value: the value is not guaranteed stable
+    Agent.Cursor: "CURSOR_AGENT",
+    # exact name, not a prefix, so GEMINI_CLI_HOME / GEMINI_CLI_NO_RELAUNCH don't match
+    Agent.GeminiCLI: "GEMINI_CLI",
+    Agent.Kiro: _is_kiro,
+    # OpenCode sets OPENCODE=1 in its root CLI middleware
+    Agent.OpenCode: "OPENCODE",
+    Agent.GitHubCopilot: "COPILOT_AGENT_SESSION_ID",
+}
+
+
+def _is_agent(agent: Agent, environ: Mapping) -> bool:
+    """
+    Check whether sam-cli is run by a particular AI agent based on certain environment variables.
+
+    Parameters
+    ----------
+    agent
+        an enum Agent object indicating which AI agent to check against.
+    environ
+        the mapping to look for environment variables, for example, os.environ.
+
+    Returns
+    -------
+    bool
+        A boolean indicating whether there are environment variables matching the agent.
+    """
+    env_var_or_callable = _ENV_VAR_OR_CALLABLE_BY_AGENT[agent]
+    if isinstance(env_var_or_callable, str):
+        return env_var_or_callable in environ
+
+    # it is a callable, use the return value
+    return env_var_or_callable(environ)
+
+
+class AgentDetector:
+    _agent: Optional[Agent]
+
+    def __init__(self):
+        try:
+            self._agent: Optional[Agent] = next(agent for agent in Agent if _is_agent(agent, os.environ))
+        except StopIteration:
+            self._agent = None
+
+    def agent(self) -> Optional[Agent]:
+        """
+        Identify which AI agent SAM CLI is running in.
+        Returns
+        -------
+        Agent
+            an optional Agent enum indicating the AI agent.
+        """
+        return self._agent
diff --git a/samcli/lib/telemetry/user_agent.py b/samcli/lib/telemetry/user_agent.py
@@ -4,10 +4,33 @@
 
 import os
 import re
-from typing import Optional
+from typing import Dict, Optional
+
+from samcli.lib.telemetry.agent_detector import Agent, AgentDetector
 
 USER_AGENT_ENV_VAR = "AWS_TOOLING_USER_AGENT"
 
+# Placeholder version emitted for every detected AI agent. Agent CLIs (e.g. Claude
+# Code) do not reliably expose their own version through the environment, so we use
+# a fixed "1.0" and rely on the agent name for grouping. Bump this only if the
+# emitted user-agent format itself needs to change.
+AGENT_USER_AGENT_VERSION = "1.0"
+
+# Stable, lowercase-with-dashes names for detected AI agents. These are combined
+# with AGENT_USER_AGENT_VERSION to build the user-agent string (e.g.
+# "claude-code/1.0") that lands in the queryable "userAgent" telemetry column, so
+# they must remain stable across releases to keep downstream queries valid. When
+# adding an agent here, also add its detection entry in agent_detector.py.
+_USER_AGENT_NAME_BY_AGENT: Dict[Agent, str] = {
+    Agent.ClaudeCode: "claude-code",
+    Agent.Codex: "codex",
+    Agent.Cursor: "cursor",
+    Agent.GeminiCLI: "gemini-cli",
+    Agent.Kiro: "kiro",
+    Agent.OpenCode: "opencode",
+    Agent.GitHubCopilot: "github-copilot",
+}
+
 # Should accept format: ${AGENT_NAME}/${AGENT_VERSION}
 # AWS_Toolkit-For-VSCode/1.62.0
 # AWS-Toolkit-For-JetBrains/1.60-223
@@ -16,7 +39,42 @@
 
 
 def get_user_agent_string() -> Optional[str]:
+    """
+    Return the user-agent string for telemetry, or None if none is available.
+
+    Precedence:
+      1. AWS_TOOLING_USER_AGENT, when set to a value matching
+         ACCEPTED_USER_AGENT_FORMAT. This is set by AWS toolkits (e.g. the VS Code
+         and JetBrains toolkits) and always wins when present, so a toolkit
+         invocation is still attributed to the toolkit even if it happens to run
+         inside an AI agent.
+      2. A detected AI agent (e.g. Claude Code), emitted as "<agent-name>/1.0".
+         This is used only as a fallback when the toolkit user-agent above is not
+         set to a valid value.
+    """
     user_agent = os.environ.get(USER_AGENT_ENV_VAR, "").strip()
     if user_agent and ACCEPTED_USER_AGENT_FORMAT.match(user_agent):
         return user_agent
+
+    return _get_agent_user_agent_string()
+
+
+def _get_agent_user_agent_string() -> Optional[str]:
+    """
+    Build a "<agent-name>/1.0" user-agent string for the detected AI agent, or None
+    when no agent is detected (or the detected agent has no configured name).
+    """
+    agent = AgentDetector().agent()
+    if agent is None:
+        return None
+
+    agent_name = _USER_AGENT_NAME_BY_AGENT.get(agent)
+    if not agent_name:
+        return None
+
+    agent_user_agent = f"{agent_name}/{AGENT_USER_AGENT_VERSION}"
+    # Validate against the same format toolkit strings must satisfy, so a
+    # misconfigured agent name never emits a malformed user-agent value.
+    if ACCEPTED_USER_AGENT_FORMAT.match(agent_user_agent):
+        return agent_user_agent
     return None
diff --git a/tests/unit/lib/telemetry/test_agent_detector.py b/tests/unit/lib/telemetry/test_agent_detector.py
@@ -0,0 +1,91 @@
+from unittest import TestCase
+from unittest.mock import patch
+
+from parameterized import parameterized
+
+from samcli.lib.telemetry.agent_detector import Agent, AgentDetector, _is_agent
+
+
+class TestAgentDetector(TestCase):
+    @parameterized.expand(
+        [
+            (Agent.ClaudeCode, "CLAUDECODE", "1"),
+            (Agent.Codex, "CODEX_SANDBOX", "seatbelt"),
+            (Agent.Cursor, "CURSOR_AGENT", "1"),
+            (Agent.GeminiCLI, "GEMINI_CLI", "1"),
+            (Agent.Kiro, "TERM_PROGRAM", "kiro"),
+            (Agent.Kiro, "AWS_EXECUTION_ENV", "AmazonQ-For-CLI Version/1.13.3"),
+            (Agent.OpenCode, "OPENCODE", "1"),
+            (Agent.GitHubCopilot, "COPILOT_AGENT_SESSION_ID", "0b8f6e2c-1a3d-4c9e-8f7a-2b1c0d9e8f7a"),
+        ]
+    )
+    def test_is_agent(self, agent, env_var, env_var_value):
+        self.assertTrue(_is_agent(agent, {env_var: env_var_value}))
+
+    @parameterized.expand(
+        [
+            (Agent.ClaudeCode, "NOT_CLAUDECODE", "1"),
+            (Agent.Codex, "NOT_CODEX_SANDBOX", "seatbelt"),
+            (Agent.Cursor, "NOT_CURSOR_AGENT", "1"),
+            # exact-name rule: GEMINI_CLI_HOME must not match Agent.GeminiCLI
+            (Agent.GeminiCLI, "GEMINI_CLI_HOME", "1"),
+            # shared-variable rule: a Lambda AWS_EXECUTION_ENV must not match Agent.Kiro
+            (Agent.Kiro, "AWS_EXECUTION_ENV", "AWS_Lambda_python3.12"),
+            # too-generic var: AGENT alone (no OPENCODE) must not match Agent.OpenCode
+            (Agent.OpenCode, "AGENT", "1"),
+        ]
+    )
+    def test_is_not_agent(self, agent, env_var, env_var_value):
+        self.assertFalse(_is_agent(agent, {env_var: env_var_value}))
+
+    @patch.dict("samcli.lib.telemetry.agent_detector.os.environ", {"CLAUDECODE": "1"}, clear=True)
+    def test_detector_identifies_claude_code(self):
+        self.assertEqual(AgentDetector().agent(), Agent.ClaudeCode)
+
+    @patch.dict("samcli.lib.telemetry.agent_detector.os.environ", {"CODEX_SANDBOX": "seatbelt"}, clear=True)
+    def test_detector_identifies_codex(self):
+        self.assertEqual(AgentDetector().agent(), Agent.Codex)
+
+    @patch.dict("samcli.lib.telemetry.agent_detector.os.environ", {"CURSOR_AGENT": "1"}, clear=True)
+    def test_detector_identifies_cursor(self):
+        self.assertEqual(AgentDetector().agent(), Agent.Cursor)
+
+    @patch.dict("samcli.lib.telemetry.agent_detector.os.environ", {"GEMINI_CLI": "1"}, clear=True)
+    def test_detector_identifies_gemini_cli(self):
+        self.assertEqual(AgentDetector().agent(), Agent.GeminiCLI)
+
+    @patch.dict("samcli.lib.telemetry.agent_detector.os.environ", {"TERM_PROGRAM": "kiro"}, clear=True)
+    def test_detector_identifies_kiro_ide(self):
+        self.assertEqual(AgentDetector().agent(), Agent.Kiro)
+
+    @patch.dict(
+        "samcli.lib.telemetry.agent_detector.os.environ",
+        {"AWS_EXECUTION_ENV": "AmazonQ-For-CLI Version/1.13.3"},
+        clear=True,
+    )
+    def test_detector_identifies_kiro_cli(self):
+        self.assertEqual(AgentDetector().agent(), Agent.Kiro)
+
+    @patch.dict(
+        "samcli.lib.telemetry.agent_detector.os.environ",
+        {"AWS_EXECUTION_ENV": "AWS_Lambda_python3.12"},
+        clear=True,
+    )
+    def test_detector_does_not_identify_lambda_as_kiro(self):
+        self.assertIsNone(AgentDetector().agent())
+
+    @patch.dict("samcli.lib.telemetry.agent_detector.os.environ", {"OPENCODE": "1"}, clear=True)
+    def test_detector_identifies_opencode(self):
+        self.assertEqual(AgentDetector().agent(), Agent.OpenCode)
+
+    @patch.dict(
+        "samcli.lib.telemetry.agent_detector.os.environ",
+        {"COPILOT_AGENT_SESSION_ID": "0b8f6e2c-1a3d-4c9e-8f7a-2b1c0d9e8f7a"},
+        clear=True,
+    )
+    def test_detector_identifies_github_copilot(self):
+        self.assertEqual(AgentDetector().agent(), Agent.GitHubCopilot)
+
+    @patch.dict("samcli.lib.telemetry.agent_detector.os.environ", {}, clear=True)
+    def test_detector_returns_none_when_no_agent(self):
+        self.assertIsNone(AgentDetector().agent())
diff --git a/tests/unit/lib/telemetry/test_event.py b/tests/unit/lib/telemetry/test_event.py
@@ -130,7 +130,12 @@ def test_track_event(self, event_mock, lock_mock):
         lock_mock.__exit__.assert_called()
 
     @patch("samcli.lib.telemetry.event.Telemetry")
-    def test_events_get_sent(self, telemetry_mock):
+    @patch("samcli.lib.telemetry.metric.get_user_agent_string")
+    def test_events_get_sent(self, get_user_agent_mock, telemetry_mock):
+        # A detected AI agent adds a "userAgent" field to every metric's common
+        # attributes; pin it to a known value so the exact-equality assertion below
+        # verifies the field is emitted rather than breaking on it.
+        get_user_agent_mock.return_value = "claude-code/1.0"
         # Create fake emit to capture tracked events
         dummy_telemetry = Mock()
         emitted_events = []
@@ -168,6 +173,7 @@ def test_events_get_sent(self, telemetry_mock):
             "samcliVersion": ANY,
             "osPlatform": ANY,
             "commandName": ANY,
+            "userAgent": "claude-code/1.0",
             "metricSpecificAttributes": {
                 "containerEngine": ANY,
                 "events": [
diff --git a/tests/unit/lib/telemetry/test_user_agent.py b/tests/unit/lib/telemetry/test_user_agent.py
@@ -3,7 +3,11 @@
 
 from parameterized import parameterized
 
-from samcli.lib.telemetry.user_agent import USER_AGENT_ENV_VAR, get_user_agent_string
+from samcli.lib.telemetry.user_agent import (
+    ACCEPTED_USER_AGENT_FORMAT,
+    USER_AGENT_ENV_VAR,
+    get_user_agent_string,
+)
 
 
 class TestUserAgent(TestCase):
@@ -43,3 +47,48 @@ def test_user_agent_without_env_var(self):
     @patch("samcli.lib.telemetry.user_agent.os.environ", {USER_AGENT_ENV_VAR: ""})
     def test_user_agent_with_empty_env_var(self):
         self.assertEqual(get_user_agent_string(), None)
+
+
+class TestUserAgentAgentFallback(TestCase):
+    # NOTE: user_agent.py and agent_detector.py both read os.environ, which is a
+    # single shared object, so one patch.dict on os.environ controls the env seen
+    # by both the toolkit lookup and the agent detector. clear=True keeps the
+    # ambient CLAUDECODE of the running shell from leaking into these tests.
+    @patch.dict("samcli.lib.telemetry.user_agent.os.environ", {"CLAUDECODE": "1"}, clear=True)
+    def test_detected_agent_is_used_when_no_toolkit_user_agent(self):
+        # No AWS_TOOLING_USER_AGENT set, Claude Code detected -> agent string is emitted.
+        self.assertEqual(get_user_agent_string(), "claude-code/1.0")
+
+    @parameterized.expand(
+        [
+            ("AWS_Toolkit-For-VSCode/1.62.0",),
+            ("AWS-Toolkit-For-JetBrains/1.60-223",),
+        ]
+    )
+    def test_toolkit_user_agent_takes_precedence_over_detected_agent(self, toolkit_user_agent):
+        # A valid AWS_TOOLING_USER_AGENT wins even when an agent is detected.
+        with patch.dict(
+            "samcli.lib.telemetry.user_agent.os.environ",
+            {"CLAUDECODE": "1", USER_AGENT_ENV_VAR: toolkit_user_agent},
+            clear=True,
+        ):
+            self.assertEqual(get_user_agent_string(), toolkit_user_agent)
+
+    @patch.dict(
+        "samcli.lib.telemetry.user_agent.os.environ",
+        {"CLAUDECODE": "1", USER_AGENT_ENV_VAR: "invalid_value"},
+        clear=True,
+    )
+    def test_detected_agent_is_used_when_toolkit_user_agent_is_invalid(self):
+        # An invalid AWS_TOOLING_USER_AGENT falls through to the detected agent.
+        self.assertEqual(get_user_agent_string(), "claude-code/1.0")
+
+    @patch.dict("samcli.lib.telemetry.user_agent.os.environ", {}, clear=True)
+    def test_none_when_no_toolkit_user_agent_and_no_agent(self):
+        self.assertEqual(get_user_agent_string(), None)
+
+    @patch.dict("samcli.lib.telemetry.user_agent.os.environ", {"CLAUDECODE": "1"}, clear=True)
+    def test_emitted_agent_string_matches_accepted_format(self):
+        # Guard against emitting a bare "ClaudeCode": the fallback must satisfy the
+        # same regex the toolkit user-agent must satisfy.
+        self.assertIsNotNone(ACCEPTED_USER_AGENT_FORMAT.match(get_user_agent_string()))
