Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 24 additions & 0 deletions stripe/_api_requestor.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
from io import BytesIO, IOBase
import json
import os
import platform
from typing import (
Any,
Expand Down Expand Up @@ -469,6 +470,23 @@ def specific_oauth_error(self, rbody, rcode, resp, rheaders, error_code):

return None

AI_AGENTS = [
("ANTIGRAVITY_CLI_ALIAS", "antigravity"),
("CLAUDECODE", "claude_code"),
("CLINE_ACTIVE", "cline"),
("CODEX_SANDBOX", "codex_cli"),
("CURSOR_AGENT", "cursor"),
("GEMINI_CLI", "gemini_cli"),
("OPENCODE", "open_code"),
]

@staticmethod
def _detect_ai_agent(environ: Mapping[str, str]) -> str:
for env_var, agent_name in _APIRequestor.AI_AGENTS:
if environ.get(env_var):
return agent_name
return ""

def request_headers(
self, method: HttpVerb, api_mode: ApiMode, options: RequestOptions
):
Expand All @@ -479,6 +497,10 @@ def request_headers(
if stripe.app_info:
user_agent += " " + self._format_app_info(stripe.app_info)

agent = self._detect_ai_agent(os.environ)
if agent:
user_agent += " AIAgent/" + agent

ua: Dict[str, Union[str, "AppInfo"]] = {
"bindings_version": VERSION,
"lang": "python",
Expand All @@ -497,6 +519,8 @@ def request_headers(
ua[attr] = val
if stripe.app_info:
ua["application"] = stripe.app_info
if agent:
ua["ai_agent"] = agent

headers: Dict[str, str] = {
"X-Stripe-Client-User-Agent": json.dumps(ua),
Expand Down
39 changes: 38 additions & 1 deletion tests/test_api_requestor.py
Original file line number Diff line number Diff line change
Expand Up @@ -671,7 +671,8 @@ def test_sets_default_http_client(self, mocker):
# the newly created client is reused
assert stripe.default_http_client == new_default_client

def test_uses_app_info(self, requestor, http_client_mock):
def test_uses_app_info(self, requestor, mocker, http_client_mock):
mocker.patch.object(_APIRequestor, "_detect_ai_agent", return_value="")
try:
old = stripe.app_info
stripe.set_app_info(
Expand Down Expand Up @@ -707,6 +708,42 @@ def test_uses_app_info(self, requestor, http_client_mock):
finally:
stripe.app_info = old

def test_detect_ai_agent(self):
assert (
_APIRequestor._detect_ai_agent({"CLAUDECODE": "1"})
== "claude_code"
)

def test_detect_ai_agent_no_env_vars(self):
assert _APIRequestor._detect_ai_agent({}) == ""

def test_detect_ai_agent_first_match_wins(self):
assert (
_APIRequestor._detect_ai_agent(
{"CURSOR_AGENT": "1", "OPENCODE": "1"}
)
== "cursor"
)

def test_ai_agent_included_in_request_headers(
self, requestor, mocker, http_client_mock
):
mocker.patch.object(
_APIRequestor, "_detect_ai_agent", return_value="cursor"
)
http_client_mock.stub_request(
"get", path=self.v1_path, rbody="{}", rcode=200
)
requestor.request("get", self.v1_path, {}, base_address="api")

last_call = http_client_mock.get_last_call()
ua = last_call.get_raw_header("User-Agent")
assert ua.endswith(" AIAgent/cursor")
client_ua = json.loads(
last_call.get_raw_header("X-Stripe-Client-User-Agent")
)
assert client_ua["ai_agent"] == "cursor"

def test_handles_failed_platform_call(
self, requestor, mocker, http_client_mock
):
Expand Down