feat: detect AI agent invocations in telemetry

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

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

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())

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": [

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()))