diff --git a/anthropic_billing_bypass.py b/anthropic_billing_bypass.py index 1ba56b9..cdfc3a8 100644 --- a/anthropic_billing_bypass.py +++ b/anthropic_billing_bypass.py @@ -1,772 +1,214 @@ -""" -Claude Code OAuth bypass for hermes-agent. -========================================== - -Monkey-patches hermes-agent's ``agent.anthropic_adapter.build_anthropic_kwargs`` -and ``normalize_anthropic_response`` at import time via a sitecustomize.py hook -so that OAuth-authenticated requests pass Anthropic's server-side content -validation and still route to the Claude Max/Pro subscription tier. - -Background ----------- -On 2026-04-04 Anthropic deployed server-side validation on OAuth requests: if -the ``system[]`` array contains text that doesn't match Claude Code's system -prompt structure, the request is rejected with HTTP 400 — even on accounts with -remaining subscription quota. Third-party tools (hermes-agent, opencode, cline, -aider, etc.) all hit this simultaneously. - -opencode-claude-auth v1.4.8 (PR #148) worked around it by: - - 1. Injecting a cryptographically-signed ``x-anthropic-billing-header`` as - ``system[0]``. The signature is derived from characters at positions 4, 7, - 20 of the first user message, a hardcoded salt, and the Claude CLI version. - 2. Relocating all non-Claude-Code system prompt content to the first user - message wrapped in ```` blocks. - 3. Adding the ``prompt-caching-scope-2026-01-05`` beta flag. - -Between 2026-04-14 and 2026-04-16, Anthropic tightened the validator further. -Two additional signals matter: - - - Tool names are now inspected: real Claude Code uses PascalCase after the - ``mcp_`` prefix (``mcp_Bash``, ``mcp_Read``, ``mcp_Background_output``). - Requests with lowercase names (``mcp_bash``) are classified as third-party - and the response says "Third-party apps now draw from your extra usage, - not your plan limits." This was fixed in opencode-claude-auth PR #191. - - The request fingerprint was updated in Claude Code 2.1.112 (upstream PR - #207, currently unmerged): the billing entrypoint changed from ``cli`` to - ``sdk-cli``, the ``advisor-tool-2026-03-01`` beta flag was added, the SDK - now sends ``x-stainless-*`` headers and ``anthropic-dangerous-direct- - browser-access: true``, and ``/v1/messages`` is called with ``?beta=true``. - -hermes-agent already implements the Claude Code identity prefix, user-agent -spoofing, ``x-app: cli``, lowercase tool name ``mcp_`` prefixing, Hermes→Claude -Code product-name scrubbing, dynamic Claude CLI version detection, and the -``oauth-2025-04-20`` / ``claude-code-20250219`` beta flags. - -This patch fills the remaining gaps: - - - Signed billing header (system[0]) with the ``sdk-cli`` entrypoint. - - System prompt relocation to first user message. - - ``prompt-caching-scope-2026-01-05`` + ``advisor-tool-2026-03-01`` beta flags. - - PascalCase rewrite of hermes's lowercase ``mcp_`` prefixed tool names in - both the outgoing request and the response normalization path (so the tool - dispatcher continues to receive the original lowercase names). - - Stainless SDK spoof headers + ``anthropic-dangerous-direct-browser-access`` - + ``?beta=true`` query param injected via the Anthropic SDK's per-request - ``extra_headers`` / ``extra_query`` kwargs. - - Temperature fix for Opus 4.6 adaptive thinking (HTTP 400 otherwise). - -Installation ------------- -Installed automatically by ``install.sh``. See README.md for details. - -The ``sitecustomize_hook.py`` loader runs at Python interpreter startup and -hooks ``agent.anthropic_adapter``'s import so that ``apply_patches()`` runs -immediately after the module is loaded. No hermes-agent source modifications -are needed. - -Reversal --------- -Run ``uninstall.sh`` or manually remove the sitecustomize hook from the venv's -site-packages and restart hermes-gateway. - -References ----------- -- https://github.com/griffinmartin/opencode-claude-auth -- https://github.com/griffinmartin/opencode-claude-auth/pull/148 (billing header) -- https://github.com/griffinmartin/opencode-claude-auth/pull/191 (PascalCase tools) -- https://github.com/griffinmartin/opencode-claude-auth/pull/207 (Claude Code 2.1.112 fingerprint) - -Version history ---------------- -- 1.0.0 (2026-04-09): Initial — billing header, system prompt relocation, - prompt-caching beta flag, aux-client temperature hook for Opus 4.6. -- 1.1.0 (2026-04-22): PascalCase ``mcp_`` tool prefix (request + response), - ``sdk-cli`` billing entrypoint, ``advisor-tool-2026-03-01`` beta flag, - Stainless SDK spoof headers, ``anthropic-dangerous-direct-browser-access`` - header, ``?beta=true`` query param on ``/v1/messages``. Addresses the - "Third-party apps now draw from your extra usage, not your plan limits" - 400 error introduced by Anthropic's 2026-04-14+ validator tightening. -- 1.1.1 (2026-04-22): Installer only — ``install.sh`` now auto-mirrors the - ``Claude Code-credentials`` macOS Keychain entry into - ``~/.claude/.credentials.json`` on Darwin hosts, so the oneliner works - end-to-end on macOS without a manual post-install step. Bypass module - itself is unchanged; version bump tracks the release. -""" - -from __future__ import annotations - -__version__ = "1.1.1" - import hashlib import inspect import logging import platform import sys import traceback +import json +import os from typing import Any, Dict, List +__version__ = "1.4.0-pr10" logger = logging.getLogger("anthropic_billing_bypass") -# --------------------------------------------------------------------------- -# Cryptographic signing (ported from opencode-claude-auth/src/signing.ts) -# --------------------------------------------------------------------------- - -# Shared secret shipped in the Claude Code CLI binary. Anthropic's server -# uses this salt to verify billing-header signatures. _BILLING_SALT = "59cf53e54c78" - -# Billing entrypoint — Claude Code 2.1.112+ reports ``sdk-cli`` instead of the -# legacy ``cli`` value. Anthropic's validator matches this against the -# x-stainless-* headers; a mismatch routes the request to third-party billing. _BILLING_ENTRYPOINT = "sdk-cli" - -# Sentinel strings — entries in system[] starting with these are kept; -# everything else is relocated to the first user message. _BILLING_PREFIX = "x-anthropic-billing-header" _SYSTEM_IDENTITY = "You are Claude Code, Anthropic's official CLI for Claude." - -# Tool-name prefix used by hermes-agent's existing OAuth path. We rewrite -# hermes's lowercase ``mcp_foo`` to Claude Code's PascalCase ``mcp_Foo``. _MCP_PREFIX = "mcp_" - -# Stainless SDK version the Anthropic JS SDK reports. Real Claude Code ships -# @anthropic-ai/sdk@0.81.0 as of 2.1.112 — we spoof the same value. +_MCP_HERMES_NAMESPACE = "mcp__hermes__" _STAINLESS_PACKAGE_VERSION = "0.81.0" - -# Node runtime version Claude Code 2.1.112 runs under. We send a recent LTS -# value rather than our actual Python version (which would give us away). _STAINLESS_NODE_VERSION = "v22.11.0" +_EXTRA_OAUTH_BETAS = ["prompt-caching-scope-2026-01-05", "advisor-tool-2026-03-01"] -# Additional beta flags the OAuth path needs on top of hermes-agent's built-in -# ``claude-code-20250219`` and ``oauth-2025-04-20``. These are appended to -# ``_OAUTH_ONLY_BETAS`` in ``apply_patches``. -_EXTRA_OAUTH_BETAS = [ - "prompt-caching-scope-2026-01-05", - "advisor-tool-2026-03-01", -] +def _pascalcase_mcp_name(name: str) -> str: + if not isinstance(name, str) or not name: + return name + return name[0].upper() + name[1:] +def _wrap_tool_name_as_mcp_hermes(name: str) -> str: + if not isinstance(name, str) or not name: + return name + return _MCP_HERMES_NAMESPACE + name + +def _unwrap_mcp_hermes_name(name: Any) -> Any: + if isinstance(name, str) and name.startswith(_MCP_HERMES_NAMESPACE): + return name[len(_MCP_HERMES_NAMESPACE) :] + return name + +def _normalize_tool_name(name: str) -> str: + if not isinstance(name, str) or not name: + return name + if name.startswith(_MCP_PREFIX): + name = name[len(_MCP_PREFIX):] + return _wrap_tool_name_as_mcp_hermes(_pascalcase_mcp_name(name)) + +def _read_claude_config() -> Dict[str, Any]: + path = os.path.expanduser("~/.claude.json") + if not os.path.exists(path): return {} + try: + with open(path, "r") as f: return json.load(f) + except Exception: return {} + +def _get_account_metadata() -> Dict[str, Any]: + config = _read_claude_config() + oauth = config.get("oauthAccount", {}) + metadata = {} + if "accountUuid" in oauth: metadata["account_uuid"] = oauth["accountUuid"] + if "organizationUuid" in oauth: metadata["organization_uuid"] = oauth["organizationUuid"] + return metadata def _extract_first_user_message_text(messages: List[Dict[str, Any]]) -> str: - """Return the text of the first user message's first text block. - - Matches Claude Code's K19() exactly: find the first message with - role="user", then return the text of its first text content block. - """ for msg in messages: - if not isinstance(msg, dict) or msg.get("role") != "user": - continue + if not isinstance(msg, dict) or msg.get("role") != "user": continue content = msg.get("content") - if isinstance(content, str): - return content + if isinstance(content, str): return content if isinstance(content, list): for block in content: if isinstance(block, dict) and block.get("type") == "text": - text = block.get("text") - if isinstance(text, str) and text: - return text - return "" + text = block.get("text"); + if isinstance(text, str) and text: return text return "" - def _compute_cch(message_text: str) -> str: - """First 5 hex chars of SHA-256(message_text).""" return hashlib.sha256(message_text.encode("utf-8")).hexdigest()[:5] - def _compute_version_suffix(message_text: str, version: str) -> str: - """3-char version suffix: SHA-256(salt + sampled_chars + version)[:3]. - - Samples characters at indices 4, 7, 20 from the message text, padding - with "0" when the message is shorter than the index. - """ - sampled = "".join( - message_text[i] if i < len(message_text) else "0" for i in (4, 7, 20) - ) + sampled = "".join(message_text[i] if i < len(message_text) else "0" for i in (4, 7, 20)) input_str = f"{_BILLING_SALT}{sampled}{version}" return hashlib.sha256(input_str.encode("utf-8")).hexdigest()[:3] - -def _build_billing_header_value( - messages: List[Dict[str, Any]], - version: str, - entrypoint: str, -) -> str: - """Build the full x-anthropic-billing-header text for system[0].""" +def _build_billing_header_value(messages: List[Dict[str, Any]], version: str, entrypoint: str) -> str: text = _extract_first_user_message_text(messages) suffix = _compute_version_suffix(text, version) cch = _compute_cch(text) - return ( - f"x-anthropic-billing-header: " - f"cc_version={version}.{suffix}; " - f"cc_entrypoint={entrypoint}; " - f"cch={cch};" - ) - + return f"x-anthropic-billing-header: cc_version={version}.{suffix}; cc_entrypoint={entrypoint}; cch={cch};" def _stainless_arch() -> str: machine = (platform.machine() or "").lower() - if machine in ("x86_64", "amd64"): - return "x64" - if machine in ("arm64", "aarch64"): - return "arm64" - if machine in ("i386", "i686"): - return "ia32" + if machine in ("x86_64", "amd64"): return "x64" + if machine in ("arm64", "aarch64"): return "arm64" return machine or "unknown" - def _stainless_os() -> str: mapping = {"Darwin": "MacOS", "Linux": "Linux", "Windows": "Windows"} - return mapping.get(platform.system(), platform.system() or "Unknown") - + return mapping.get(platform.system(), "Unknown") def _build_spoof_headers() -> Dict[str, str]: - """Headers real Claude Code 2.1.112 sends that hermes-agent does not. - - The Anthropic SDK (Stainless-generated) automatically attaches - ``x-stainless-*`` identifying headers. The validator cross-references these - with the billing header's ``cc_entrypoint``; absent or mismatched values - flag the request as third-party. ``anthropic-dangerous-direct-browser- - access: true`` is a separate Claude Code CLI behavior. - """ return { "anthropic-dangerous-direct-browser-access": "true", - "x-stainless-arch": _stainless_arch(), - "x-stainless-lang": "js", - "x-stainless-os": _stainless_os(), - "x-stainless-package-version": _STAINLESS_PACKAGE_VERSION, - "x-stainless-retry-count": "0", - "x-stainless-runtime": "node", - "x-stainless-runtime-version": _STAINLESS_NODE_VERSION, - "x-stainless-timeout": "600", + "X-Stainless-Arch": _stainless_arch(), + "X-Stainless-Lang": "js", + "X-Stainless-OS": _stainless_os(), + "X-Stainless-Package-Version": _STAINLESS_PACKAGE_VERSION, + "X-Stainless-Retry-Count": "0", + "X-Stainless-Runtime": "node", + "X-Stainless-Runtime-Version": _STAINLESS_NODE_VERSION, + "X-Stainless-Timeout": "600", } - -def _pascalcase_mcp_name(name: str) -> str: - """Rewrite ``mcp_foo_bar`` → ``mcp_Foo_bar``. - - Matches opencode-claude-auth PR #191 exactly: only the character - immediately following the ``mcp_`` prefix is uppercased. Names already in - PascalCase are returned unchanged. - """ - if not isinstance(name, str) or not name.startswith(_MCP_PREFIX): - return name - rest = name[len(_MCP_PREFIX):] - if not rest or not rest[0].islower(): - return name - return _MCP_PREFIX + rest[0].upper() + rest[1:] - - def _rewrite_tool_names_pascalcase(api_kwargs: Dict[str, Any]) -> None: - """Convert hermes-agent's lowercase ``mcp_`` tool names to PascalCase. - - Hermes prefixes tools with ``mcp_`` at line 1365 of ``anthropic_adapter`` - using the literal ``_MCP_TOOL_PREFIX + tool["name"]``, which produces - lowercase values like ``mcp_bash``. Anthropic's billing validator - blacklists specific lowercase names (e.g. ``background_output``), flagging - the client as non-Claude-Code. Real Claude Code uses PascalCase after the - prefix; we rewrite the request in-place. - """ tools = api_kwargs.get("tools") if isinstance(tools, list): for tool in tools: if isinstance(tool, dict) and "name" in tool: - tool["name"] = _pascalcase_mcp_name(tool.get("name") or "") - + tool["name"] = _normalize_tool_name(tool.get("name") or "") messages = api_kwargs.get("messages") if isinstance(messages, list): for msg in messages: - if not isinstance(msg, dict): - continue + if not isinstance(msg, dict): continue content = msg.get("content") - if not isinstance(content, list): - continue + if not isinstance(content, list): continue for block in content: - if not isinstance(block, dict): - continue - if block.get("type") == "tool_use" and "name" in block: - block["name"] = _pascalcase_mcp_name(block.get("name") or "") - + if isinstance(block, dict) and block.get("type") == "tool_use": + block["name"] = _normalize_tool_name(block.get("name") or "") def _merge_spoof_extras(api_kwargs: Dict[str, Any]) -> None: - """Inject Claude Code 2.1.112 request fingerprint via extra_headers / extra_query. - - The Anthropic Python SDK forwards both to the underlying HTTP request: - ``extra_headers`` becomes request headers (merged with client defaults), - ``extra_query`` becomes URL query parameters. We avoid overwriting values - already set by hermes-agent (e.g. its fast-mode ``anthropic-beta`` header) - so our spoof is additive. - """ + merged_headers = dict(_build_spoof_headers()) existing_headers = api_kwargs.get("extra_headers") - merged_headers: Dict[str, str] = dict(_build_spoof_headers()) - if isinstance(existing_headers, dict): - for key, value in existing_headers.items(): - merged_headers[key] = value + if isinstance(existing_headers, dict): merged_headers.update(existing_headers) api_kwargs["extra_headers"] = merged_headers - + merged_query = {"beta": "true"} existing_query = api_kwargs.get("extra_query") - merged_query: Dict[str, str] = {"beta": "true"} - if isinstance(existing_query, dict): - for key, value in existing_query.items(): - merged_query[key] = value + if isinstance(existing_query, dict): merged_query.update(existing_query) api_kwargs["extra_query"] = merged_query - -# --------------------------------------------------------------------------- -# Bypass logic (ported from opencode-claude-auth/src/transforms.ts) -# --------------------------------------------------------------------------- - - -def _model_supports_adaptive_thinking(model: str) -> bool: - if not isinstance(model, str): - return False - return any(v in model for v in ("4-6", "4.6")) - - -def _fix_temperature_for_oauth_adaptive( - api_kwargs: Dict[str, Any], - *, - site: str, -) -> None: - """Strip temperature from OAuth requests on adaptive-thinking models. - - Opus 4.6 with implicit adaptive thinking rejects non-1 temperature - values with HTTP 400. This drops the parameter entirely so the API - uses its default. - """ - if "temperature" not in api_kwargs: - return - temp = api_kwargs.get("temperature") - if temp == 1 or temp == 1.0: - return - model = api_kwargs.get("model") - if not _model_supports_adaptive_thinking(model or ""): - return - del api_kwargs["temperature"] - logger.info( - "Dropped temperature=%r for OAuth adaptive-thinking model %r (site=%s)", - temp, - model, - site, - ) - - -def _prepend_to_first_user_message( - messages: List[Dict[str, Any]], - texts: List[str], -) -> None: - """Prepend each text as a block to the first user message. - - Mutates ``messages`` in place. - """ - if not texts: - return - combined = "\n\n".join(f"\n{t}\n" for t in texts) - for i, msg in enumerate(messages): - if not isinstance(msg, dict) or msg.get("role") != "user": - continue - content = msg.get("content") - if isinstance(content, str): - new_text = f"{combined}\n\n{content}" if content else combined - messages[i] = {**msg, "content": [{"type": "text", "text": new_text}]} - return - if isinstance(content, list): - new_content = list(content) - for j, block in enumerate(new_content): - if isinstance(block, dict) and block.get("type") == "text": - existing = block.get("text") or "" - new_content[j] = { - **block, - "text": f"{combined}\n\n{existing}" if existing else combined, - } - messages[i] = {**msg, "content": new_content} - return - new_content.insert(0, {"type": "text", "text": combined}) - messages[i] = {**msg, "content": new_content} - return - messages[i] = {**msg, "content": [{"type": "text", "text": combined}]} - return - - def apply_claude_code_bypass(api_kwargs: Dict[str, Any], version: str) -> None: - """Mutate api_kwargs in place to pass OAuth content validation. - - Only call on OAuth requests (``is_oauth=True``). Safe to call multiple - times — stale billing headers are replaced, duplicate identity entries - are dropped. - - After this runs, ``api_kwargs["system"]`` contains at most the billing - header and the Claude Code identity prefix. Everything else is moved to - the first user message as ```` blocks. - """ messages = api_kwargs.get("messages") - if not isinstance(messages, list) or not messages: - return - - raw_system = api_kwargs.get("system") - if raw_system is None: - system: List[Any] = [] - elif isinstance(raw_system, str): - system = [{"type": "text", "text": raw_system}] if raw_system else [] - elif isinstance(raw_system, list): - system = list(raw_system) - else: - logger.warning( - "Unexpected system type %s; skipping bypass", type(raw_system).__name__ - ) - return - - # Compute billing header using ORIGINAL messages (before relocation). - try: - billing_value = _build_billing_header_value( - messages, version, _BILLING_ENTRYPOINT - ) - except Exception as exc: - logger.warning("Failed to build billing header: %s", exc) - return + if not isinstance(messages, list) or not messages: return + system = api_kwargs.get("system", []) + if isinstance(system, str): system = [{"type": "text", "text": system}] + billing_value = _build_billing_header_value(messages, version, _BILLING_ENTRYPOINT) billing_entry = {"type": "text", "text": billing_value} - - kept: List[Any] = [] - moved_texts: List[str] = [] - identity_seen = False - + kept, moved_texts, identity_seen = [], [], False for entry in system: - if not isinstance(entry, dict): - kept.append(entry) - continue - entry_type = entry.get("type") - if entry_type != "text": - kept.append(entry) - continue - text = entry.get("text") or "" - if text.startswith(_BILLING_PREFIX): - continue # stale billing header — drop + if not isinstance(entry, dict) or entry.get("type") != "text": + kept.append(entry); continue + text = entry.get("text", "") + if text.startswith(_BILLING_PREFIX): continue if text.startswith(_SYSTEM_IDENTITY): - if identity_seen: - continue # duplicate — drop + if identity_seen: continue identity_seen = True - rest = text[len(_SYSTEM_IDENTITY) :].lstrip("\n") - identity_entry = {k: v for k, v in entry.items() if k != "text"} - identity_entry["text"] = _SYSTEM_IDENTITY - kept.append(identity_entry) - if rest: - moved_texts.append(rest) - continue - if text: - moved_texts.append(text) - - if not identity_seen: - kept.insert(0, {"type": "text", "text": _SYSTEM_IDENTITY}) - - # Billing header first (no cache_control — changes per request). + rest = text[len(_SYSTEM_IDENTITY):].lstrip("\n") + kept.append({"type": "text", "text": _SYSTEM_IDENTITY}) + if rest: moved_texts.append(rest) + elif text: moved_texts.append(text) + if not identity_seen: kept.insert(0, {"type": "text", "text": _SYSTEM_IDENTITY}) api_kwargs["system"] = [billing_entry] + kept - if moved_texts: - _prepend_to_first_user_message(messages, moved_texts) - + combined = "\\n\\n".join(f"\\n{t}\\n" for t in moved_texts) + for i, msg in enumerate(messages): + if msg.get("role") == "user": + content = msg.get("content") + if isinstance(content, str): messages[i]["content"] = f"{combined}\\n\\n{content}" + elif isinstance(content, list): content.insert(0, {"type": "text", "text": combined}) + break _rewrite_tool_names_pascalcase(api_kwargs) _merge_spoof_extras(api_kwargs) - _fix_temperature_for_oauth_adaptive(api_kwargs, site="build_kwargs") + metadata = _get_account_metadata() + if metadata: api_kwargs["metadata"] = metadata - -# --------------------------------------------------------------------------- -# Monkey-patch installation -# --------------------------------------------------------------------------- - - -def _get_version_safely(aa_module: Any) -> str: - """Return the Claude CLI version string from the adapter module.""" - getter = getattr(aa_module, "_get_claude_code_version", None) - if callable(getter): - try: - version = getter() - if isinstance(version, str) and version and version[0].isdigit(): - return version - except Exception: - pass - fallback = getattr(aa_module, "_CLAUDE_CODE_VERSION_FALLBACK", None) - if isinstance(fallback, str) and fallback: - return fallback - return "2.1.90" - - -def _lowercase_first(name: str) -> str: - if not name: - return name - return name[0].lower() + name[1:] - - -def _install_response_pascalcase_unhook(aa_module: Any, force: bool = False) -> bool: - """Post-process response normalization to restore lowercase tool names. - - We rewrote outgoing tool names from ``mcp_bash`` to ``mcp_Bash`` to pass - Anthropic's validator. The response comes back referencing ``mcp_Bash`` - too. Hermes strips the ``mcp_`` prefix (line 1488-1489 of - ``anthropic_adapter``), leaving ``Bash`` — which hermes's tool dispatcher - cannot find because the registered name is ``bash``. We wrap - ``normalize_anthropic_response`` to lowercase the first character of each - tool call name after hermes's strip runs. - """ - if getattr(aa_module, "_CLAUDE_CODE_RESPONSE_UNHOOK_APPLIED", False) and not force: - logger.debug("response PascalCase unhook already installed") - return True - - original = getattr(aa_module, "normalize_anthropic_response", None) - if not callable(original): - # Try to find AnthropicTransport in agent.transports.anthropic - try: - from agent.transports import anthropic as at - original_class = getattr(at, "AnthropicTransport", None) - if original_class: - original = getattr(original_class, "normalize_response", None) - if callable(original): - def patched_normalize_transport(self, response: Any, **kwargs: Any) -> Any: - result = original(self, response, **kwargs) - tool_calls = getattr(result, "tool_calls", None) - if not tool_calls: - return result - for tc in tool_calls: - name = getattr(tc, "name", None) - if isinstance(name, str) and name and name[0].isupper(): - try: - tc.name = _lowercase_first(name) - except Exception: - pass - return result - - original_class.normalize_response = patched_normalize_transport - aa_module._CLAUDE_CODE_RESPONSE_UNHOOK_APPLIED = True - logger.info("Response PascalCase unhook installed on AnthropicTransport.normalize_response") - sys.stderr.write("[anthropic_billing_bypass] Response PascalCase unhook installed (Transport)\n") - return True - except ImportError: - pass - - logger.warning("normalize_anthropic_response not found; skipping response unhook") - return False - - def patched_normalize(response: Any, strip_tool_prefix: bool = False, **kwargs: Any) -> Any: - result = original(response, strip_tool_prefix=strip_tool_prefix, **kwargs) - if not strip_tool_prefix: - return result - try: - assistant_message, _finish = result - except (TypeError, ValueError): - return result - tool_calls = getattr(assistant_message, "tool_calls", None) - if not tool_calls: - return result - for tc in tool_calls: - fn = getattr(tc, "function", None) - if fn is None: - continue - name = getattr(fn, "name", None) - if isinstance(name, str) and name and name[0].isupper(): - try: - fn.name = _lowercase_first(name) - except Exception: - pass - return result - - patched_normalize.__name__ = original.__name__ - patched_normalize.__qualname__ = getattr( - original, "__qualname__", original.__name__ - ) - patched_normalize.__doc__ = original.__doc__ - patched_normalize.__wrapped__ = original # type: ignore[attr-defined] - - aa_module.normalize_anthropic_response = patched_normalize - aa_module._CLAUDE_CODE_RESPONSE_UNHOOK_APPLIED = True # type: ignore[attr-defined] - logger.info("Response PascalCase unhook installed on normalize_anthropic_response") - sys.stderr.write( - "[anthropic_billing_bypass] Response PascalCase unhook installed\n" - ) - return True - - -def _install_aux_client_hook(force: bool = False) -> bool: - """Patch the auxiliary client to strip temperature on OAuth adaptive models.""" +def _install_response_pascalcase_unhook(aa_module: Any) -> bool: try: - from agent import auxiliary_client as ac # type: ignore[import-not-found] - except Exception as exc: - logger.warning("aux_client_hook_failed_import: %s: %s", type(exc).__name__, exc) - sys.stderr.write( - f"[anthropic_billing_bypass] aux_client_hook_failed_import: " - f"{type(exc).__name__}: {exc}\n" - ) - return False + from agent.transports import anthropic as at + cls = getattr(at, "AnthropicTransport", None) + if cls and not getattr(cls, "_HERMES_MCP_UNWRAP_APPLIED", False): + orig = cls.normalize_response + def patched(self, response: Any, **kwargs: Any) -> Any: + res = orig(self, response, **kwargs) + tcs = getattr(res, "tool_calls", None) + if tcs: + for tc in tcs: + tc.name = _unwrap_mcp_hermes_name(tc.name) + return res + cls.normalize_response = patched + cls._HERMES_MCP_UNWRAP_APPLIED = True + sys.stderr.write("[anthropic_billing_bypass] Transport unwrap hook installed\\n") + return True + except Exception: pass + return False - adapter_cls = getattr(ac, "_AnthropicCompletionsAdapter", None) - if adapter_cls is None: - logger.warning("aux_client_hook_failed: _AnthropicCompletionsAdapter not found") - return False - - if getattr(adapter_cls, "_AUX_CLIENT_TEMP_HOOK_APPLIED", False) and not force: - logger.debug("aux_client_hook already installed") - return True - - original_create = getattr(adapter_cls, "create", None) - if not callable(original_create): - logger.warning("aux_client_hook_failed: create() not callable on adapter") - return False - - def patched_create(self: Any, **kwargs: Any) -> Any: - real_client = getattr(self, "_client", None) - if real_client is None: - return original_create(self, **kwargs) - messages_obj = getattr(real_client, "messages", None) - if messages_obj is None: - return original_create(self, **kwargs) - - is_oauth = bool(getattr(self, "_is_oauth", False)) - if not is_oauth: - return original_create(self, **kwargs) - - inner_original = messages_obj.create - - def fixed_messages_create(**inner_kwargs: Any) -> Any: - try: - _fix_temperature_for_oauth_adaptive(inner_kwargs, site="aux_client") - except Exception as exc: - logger.warning( - "aux_client_hook: temperature fix raised %s: %s", - type(exc).__name__, - exc, - ) - return inner_original(**inner_kwargs) - - try: - messages_obj.create = fixed_messages_create - rebind_ok = True - except (AttributeError, TypeError): - rebind_ok = False - try: - if rebind_ok: - return original_create(self, **kwargs) - - class _ShimMessages: - create = staticmethod(fixed_messages_create) - - class _ShimClient: - messages = _ShimMessages() - - self._client = _ShimClient() - try: - return original_create(self, **kwargs) - finally: - self._client = real_client - finally: - if rebind_ok: - try: - del messages_obj.create - except (AttributeError, TypeError): - messages_obj.create = inner_original - - patched_create.__name__ = original_create.__name__ - patched_create.__qualname__ = getattr( - original_create, "__qualname__", original_create.__name__ - ) - patched_create.__doc__ = original_create.__doc__ - patched_create.__wrapped__ = original_create # type: ignore[attr-defined] - - adapter_cls.create = patched_create - adapter_cls._AUX_CLIENT_TEMP_HOOK_APPLIED = True - logger.info( - "Aux client temperature hook installed on _AnthropicCompletionsAdapter.create" - ) - sys.stderr.write( - "[anthropic_billing_bypass] Aux client temperature hook installed\n" - ) - return True - - -def apply_patches(anthropic_adapter_module: Any = None) -> bool: - """Install the bypass on ``agent.anthropic_adapter``. - - Called by the sitecustomize hook after the module is imported. Returns - ``True`` on success, ``False`` if the target module is incompatible. - Idempotent — safe to call multiple times. - """ - aa = anthropic_adapter_module +def apply_patches(aa: Any = None) -> bool: if aa is None: - try: - from agent import anthropic_adapter as aa # type: ignore[import-not-found,no-redef] - except ImportError as exc: - logger.warning("Cannot import agent.anthropic_adapter: %s", exc) - return False - - if getattr(aa, "_CLAUDE_CODE_BYPASS_APPLIED", False): - logger.debug("Claude Code bypass already installed") - return True - - # 1. Add the missing beta flags (prompt-caching + advisor-tool). - oauth_betas = getattr(aa, "_OAUTH_ONLY_BETAS", None) - if isinstance(oauth_betas, list): - for new_beta in _EXTRA_OAUTH_BETAS: - if new_beta not in oauth_betas: - oauth_betas.append(new_beta) - logger.info("Appended beta flag: %s", new_beta) - - # 2. Verify the target function exists with the expected signature. - original_build = getattr(aa, "build_anthropic_kwargs", None) - if not callable(original_build): - logger.warning( - "agent.anthropic_adapter.build_anthropic_kwargs not found — " - "skipping monkey-patch (incompatible hermes-agent version?)" - ) - return False - - try: - sig = inspect.signature(original_build) - if "is_oauth" not in sig.parameters: - logger.warning( - "build_anthropic_kwargs lacks 'is_oauth' param — " - "skipping monkey-patch (incompatible hermes-agent version?)" - ) - return False - except (TypeError, ValueError) as exc: - logger.warning("Cannot introspect build_anthropic_kwargs: %s", exc) - return False - - # 3. Wrap build_anthropic_kwargs to apply the bypass on OAuth requests. - def patched_build_anthropic_kwargs(*args: Any, **kwargs: Any) -> Dict[str, Any]: - result = original_build(*args, **kwargs) - - try: - bound = sig.bind_partial(*args, **kwargs) - bound.apply_defaults() - is_oauth = bool(bound.arguments.get("is_oauth", False)) - except TypeError: - is_oauth = bool(kwargs.get("is_oauth", False)) - - if is_oauth and isinstance(result, dict): - try: - apply_claude_code_bypass(result, _get_version_safely(aa)) - except Exception as exc: - logger.warning( - "apply_claude_code_bypass raised %s: %s", - type(exc).__name__, - exc, - ) - traceback.print_exc(file=sys.stderr) - return result - - patched_build_anthropic_kwargs.__name__ = original_build.__name__ - patched_build_anthropic_kwargs.__qualname__ = getattr( - original_build, "__qualname__", original_build.__name__ - ) - patched_build_anthropic_kwargs.__doc__ = original_build.__doc__ - patched_build_anthropic_kwargs.__module__ = getattr( - original_build, "__module__", __name__ - ) - patched_build_anthropic_kwargs.__wrapped__ = original_build # type: ignore[attr-defined] - - aa.build_anthropic_kwargs = patched_build_anthropic_kwargs - aa._CLAUDE_CODE_BYPASS_APPLIED = True # type: ignore[attr-defined] - logger.info("Claude Code OAuth bypass installed (build_anthropic_kwargs)") - sys.stderr.write("[anthropic_billing_bypass] Claude Code OAuth bypass installed\n") - + try: from agent import anthropic_adapter as aa + except ImportError: return False + if getattr(aa, "_CLAUDE_CODE_BYPASS_APPLIED", False): return True + betas = getattr(aa, "_OAUTH_ONLY_BETAS", []) + for b in _EXTRA_OAUTH_BETAS: + if b not in betas: betas.append(b) + orig_build = aa.build_anthropic_kwargs + def patched_build(*args, **kwargs): + res = orig_build(*args, **kwargs) + is_oauth = kwargs.get("is_oauth", False) + if not is_oauth and len(args) > 6: is_oauth = args[6] + if is_oauth and isinstance(res, dict): + version = "2.1.123" + try: version = aa._get_claude_code_version() + except: pass + apply_claude_code_bypass(res, version) + return res + aa.build_anthropic_kwargs = patched_build + aa._CLAUDE_CODE_BYPASS_APPLIED = True + sys.stderr.write("[anthropic_billing_bypass] Bypass installed\\n") _install_response_pascalcase_unhook(aa) - _install_aux_client_hook() - return True