public/hermes-claude-auth
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update bypass with advanced PR #10 logic (namespacing, account linking, telemetry stripping)

Browse Source
This commit is contained in:
graydeon 2026-04-29 13:02:06 -05:00
parent d834eb67aa
commit 5e35bad69f
1 changed files with 130 additions and 688 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

818
anthropic_billing_bypass.py
View File

@ -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 ``<system-reminder>`` 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 hashlib
import inspect import inspect
import logging import logging
import platform import platform
import sys import sys
import traceback import traceback
import json
import os
from typing import Any, Dict, List from typing import Any, Dict, List
__version__ = "1.4.0-pr10"
logger = logging.getLogger("anthropic_billing_bypass") 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_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" _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" _BILLING_PREFIX = "x-anthropic-billing-header"
_SYSTEM_IDENTITY = "You are Claude Code, Anthropic's official CLI for Claude." _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_" _MCP_PREFIX = "mcp_"
_MCP_HERMES_NAMESPACE = "mcp__hermes__"
# 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.
_STAINLESS_PACKAGE_VERSION = "0.81.0" _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" _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 def _pascalcase_mcp_name(name: str) -> str:
# ``claude-code-20250219`` and ``oauth-2025-04-20``. These are appended to if not isinstance(name, str) or not name:
# ``_OAUTH_ONLY_BETAS`` in ``apply_patches``. return name
_EXTRA_OAUTH_BETAS = [ return name[0].upper() + name[1:]
"prompt-caching-scope-2026-01-05",
"advisor-tool-2026-03-01",
]
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: 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: for msg in messages:
if not isinstance(msg, dict) or msg.get("role") != "user": if not isinstance(msg, dict) or msg.get("role") != "user": continue
continue
content = msg.get("content") content = msg.get("content")
if isinstance(content, str): if isinstance(content, str): return content
return content
if isinstance(content, list): if isinstance(content, list):
for block in content: for block in content:
if isinstance(block, dict) and block.get("type") == "text": if isinstance(block, dict) and block.get("type") == "text":
text = block.get("text") text = block.get("text");
if isinstance(text, str) and text: if isinstance(text, str) and text: return text
return text
return ""
return "" return ""
def _compute_cch(message_text: str) -> str: 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] return hashlib.sha256(message_text.encode("utf-8")).hexdigest()[:5]
def _compute_version_suffix(message_text: str, version: str) -> str: def _compute_version_suffix(message_text: str, version: str) -> str:
"""3-char version suffix: SHA-256(salt + sampled_chars + version)[:3]. sampled = "".join(message_text[i] if i < len(message_text) else "0" for i in (4, 7, 20))
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)
)
input_str = f"{_BILLING_SALT}{sampled}{version}" input_str = f"{_BILLING_SALT}{sampled}{version}"
return hashlib.sha256(input_str.encode("utf-8")).hexdigest()[:3] 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:
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]."""
text = _extract_first_user_message_text(messages) text = _extract_first_user_message_text(messages)
suffix = _compute_version_suffix(text, version) suffix = _compute_version_suffix(text, version)
cch = _compute_cch(text) cch = _compute_cch(text)
return ( return f"x-anthropic-billing-header: cc_version={version}.{suffix}; cc_entrypoint={entrypoint}; cch={cch};"
f"x-anthropic-billing-header: "
f"cc_version={version}.{suffix}; "
f"cc_entrypoint={entrypoint}; "
f"cch={cch};"
)
def _stainless_arch() -> str: def _stainless_arch() -> str:
machine = (platform.machine() or "").lower() machine = (platform.machine() or "").lower()
if machine in ("x86_64", "amd64"): if machine in ("x86_64", "amd64"): return "x64"
return "x64" if machine in ("arm64", "aarch64"): return "arm64"
if machine in ("arm64", "aarch64"):
return "arm64"
if machine in ("i386", "i686"):
return "ia32"
return machine or "unknown" return machine or "unknown"
def _stainless_os() -> str: def _stainless_os() -> str:
mapping = {"Darwin": "MacOS", "Linux": "Linux", "Windows": "Windows"} 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]: 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 { return {
"anthropic-dangerous-direct-browser-access": "true", "anthropic-dangerous-direct-browser-access": "true",
"x-stainless-arch": _stainless_arch(), "X-Stainless-Arch": _stainless_arch(),
"x-stainless-lang": "js", "X-Stainless-Lang": "js",
"x-stainless-os": _stainless_os(), "X-Stainless-OS": _stainless_os(),
"x-stainless-package-version": _STAINLESS_PACKAGE_VERSION, "X-Stainless-Package-Version": _STAINLESS_PACKAGE_VERSION,
"x-stainless-retry-count": "0", "X-Stainless-Retry-Count": "0",
"x-stainless-runtime": "node", "X-Stainless-Runtime": "node",
"x-stainless-runtime-version": _STAINLESS_NODE_VERSION, "X-Stainless-Runtime-Version": _STAINLESS_NODE_VERSION,
"x-stainless-timeout": "600", "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: 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") tools = api_kwargs.get("tools")
if isinstance(tools, list): if isinstance(tools, list):
for tool in tools: for tool in tools:
if isinstance(tool, dict) and "name" in tool: 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") messages = api_kwargs.get("messages")
if isinstance(messages, list): if isinstance(messages, list):
for msg in messages: for msg in messages:
if not isinstance(msg, dict): if not isinstance(msg, dict): continue
continue
content = msg.get("content") content = msg.get("content")
if not isinstance(content, list): if not isinstance(content, list): continue
continue
for block in content: for block in content:
if not isinstance(block, dict): if isinstance(block, dict) and block.get("type") == "tool_use":
continue block["name"] = _normalize_tool_name(block.get("name") or "")
if block.get("type") == "tool_use" and "name" in block:
block["name"] = _pascalcase_mcp_name(block.get("name") or "")
def _merge_spoof_extras(api_kwargs: Dict[str, Any]) -> None: def _merge_spoof_extras(api_kwargs: Dict[str, Any]) -> None:
"""Inject Claude Code 2.1.112 request fingerprint via extra_headers / extra_query. merged_headers = dict(_build_spoof_headers())
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.
"""
existing_headers = api_kwargs.get("extra_headers") existing_headers = api_kwargs.get("extra_headers")
merged_headers: Dict[str, str] = dict(_build_spoof_headers()) if isinstance(existing_headers, dict): merged_headers.update(existing_headers)
if isinstance(existing_headers, dict):
for key, value in existing_headers.items():
merged_headers[key] = value
api_kwargs["extra_headers"] = merged_headers api_kwargs["extra_headers"] = merged_headers
merged_query = {"beta": "true"}
existing_query = api_kwargs.get("extra_query") existing_query = api_kwargs.get("extra_query")
merged_query: Dict[str, str] = {"beta": "true"} if isinstance(existing_query, dict): merged_query.update(existing_query)
if isinstance(existing_query, dict):
for key, value in existing_query.items():
merged_query[key] = value
api_kwargs["extra_query"] = merged_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 <system-reminder> block to the first user message.
Mutates ``messages`` in place.
"""
if not texts:
return
combined = "\n\n".join(f"<system-reminder>\n{t}\n</system-reminder>" 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: 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 ``<system-reminder>`` blocks.
"""
messages = api_kwargs.get("messages") messages = api_kwargs.get("messages")
if not isinstance(messages, list) or not messages: if not isinstance(messages, list) or not messages: return
return system = api_kwargs.get("system", [])
if isinstance(system, str): system = [{"type": "text", "text": system}]
raw_system = api_kwargs.get("system") billing_value = _build_billing_header_value(messages, version, _BILLING_ENTRYPOINT)
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
billing_entry = {"type": "text", "text": billing_value} billing_entry = {"type": "text", "text": billing_value}
kept, moved_texts, identity_seen = [], [], False
kept: List[Any] = []
moved_texts: List[str] = []
identity_seen = False
for entry in system: for entry in system:
if not isinstance(entry, dict): if not isinstance(entry, dict) or entry.get("type") != "text":
kept.append(entry) kept.append(entry); continue
continue text = entry.get("text", "")
entry_type = entry.get("type") if text.startswith(_BILLING_PREFIX): continue
if entry_type != "text":
kept.append(entry)
continue
text = entry.get("text") or ""
if text.startswith(_BILLING_PREFIX):
continue # stale billing header — drop
if text.startswith(_SYSTEM_IDENTITY): if text.startswith(_SYSTEM_IDENTITY):
if identity_seen: if identity_seen: continue
continue # duplicate — drop
identity_seen = True identity_seen = True
rest = text[len(_SYSTEM_IDENTITY) :].lstrip("\n") rest = text[len(_SYSTEM_IDENTITY):].lstrip("\n")
identity_entry = {k: v for k, v in entry.items() if k != "text"} kept.append({"type": "text", "text": _SYSTEM_IDENTITY})
identity_entry["text"] = _SYSTEM_IDENTITY if rest: moved_texts.append(rest)
kept.append(identity_entry) elif text: moved_texts.append(text)
if rest: if not identity_seen: kept.insert(0, {"type": "text", "text": _SYSTEM_IDENTITY})
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).
api_kwargs["system"] = [billing_entry] + kept api_kwargs["system"] = [billing_entry] + kept
if moved_texts: if moved_texts:
_prepend_to_first_user_message(messages, moved_texts) combined = "\\n\\n".join(f"<system-reminder>\\n{t}\\n</system-reminder>" 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) _rewrite_tool_names_pascalcase(api_kwargs)
_merge_spoof_extras(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
def _install_response_pascalcase_unhook(aa_module: Any) -> bool:
# ---------------------------------------------------------------------------
# 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."""
try: try:
from agent import auxiliary_client as ac # type: ignore[import-not-found] from agent.transports import anthropic as at
except Exception as exc: cls = getattr(at, "AnthropicTransport", None)
logger.warning("aux_client_hook_failed_import: %s: %s", type(exc).__name__, exc) if cls and not getattr(cls, "_HERMES_MCP_UNWRAP_APPLIED", False):
sys.stderr.write( orig = cls.normalize_response
f"[anthropic_billing_bypass] aux_client_hook_failed_import: " def patched(self, response: Any, **kwargs: Any) -> Any:
f"{type(exc).__name__}: {exc}\n" res = orig(self, response, **kwargs)
) tcs = getattr(res, "tool_calls", None)
return False 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) def apply_patches(aa: Any = None) -> bool:
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
if aa is None: if aa is None:
try: try: from agent import anthropic_adapter as aa
from agent import anthropic_adapter as aa # type: ignore[import-not-found,no-redef] except ImportError: return False
except ImportError as exc: if getattr(aa, "_CLAUDE_CODE_BYPASS_APPLIED", False): return True
logger.warning("Cannot import agent.anthropic_adapter: %s", exc) betas = getattr(aa, "_OAUTH_ONLY_BETAS", [])
return False for b in _EXTRA_OAUTH_BETAS:
if b not in betas: betas.append(b)
if getattr(aa, "_CLAUDE_CODE_BYPASS_APPLIED", False): orig_build = aa.build_anthropic_kwargs
logger.debug("Claude Code bypass already installed") def patched_build(*args, **kwargs):
return True res = orig_build(*args, **kwargs)
is_oauth = kwargs.get("is_oauth", False)
# 1. Add the missing beta flags (prompt-caching + advisor-tool). if not is_oauth and len(args) > 6: is_oauth = args[6]
oauth_betas = getattr(aa, "_OAUTH_ONLY_BETAS", None) if is_oauth and isinstance(res, dict):
if isinstance(oauth_betas, list): version = "2.1.123"
for new_beta in _EXTRA_OAUTH_BETAS: try: version = aa._get_claude_code_version()
if new_beta not in oauth_betas: except: pass
oauth_betas.append(new_beta) apply_claude_code_bypass(res, version)
logger.info("Appended beta flag: %s", new_beta) return res
aa.build_anthropic_kwargs = patched_build
# 2. Verify the target function exists with the expected signature. aa._CLAUDE_CODE_BYPASS_APPLIED = True
original_build = getattr(aa, "build_anthropic_kwargs", None) sys.stderr.write("[anthropic_billing_bypass] Bypass installed\\n")
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")
_install_response_pascalcase_unhook(aa) _install_response_pascalcase_unhook(aa)
_install_aux_client_hook()
return True return True