fix(anthropic-oauth): bypass Anthropic's spoof filter for Claude subscription auth

agent/anthropic_adapter.py

@@ -311,7 +311,10 @@ def _detect_claude_code_version() -> str:
 
 
 _CLAUDE_CODE_SYSTEM_PREFIX = "You are Claude Code, Anthropic's official CLI for Claude."
-_MCP_TOOL_PREFIX = "mcp_"
+# Anthropic's OAuth/subscription filter rejects single-underscore `mcp_*` tool
+# names (treats them as spoofing the MCP namespace). Claude Code's real MCP
+# convention is `mcp__<server>__<tool>` with double underscores.
+_MCP_TOOL_PREFIX = "mcp__hermes__"
 
 
 def _get_claude_code_version() -> str:
@@ -1933,44 +1936,74 @@ def build_anthropic_kwargs(
         effective_max_tokens = max(context_length - 1, 1)
 
     # ── OAuth: Claude Code identity ──────────────────────────────────
+    # Anthropic's subscription OAuth path enforces a strict spoof filter:
+    #   * System prompt: effectively only the official Claude Code prefix is
+    #     allowed. Extra system text — even after replacing "Hermes Agent" →
+    #     "Claude Code" — trips a 400 "out of extra usage" rejection once
+    #     it crosses ~2 KB or mentions Claude Code in non-official ways.
+    #   * Tool names: must use Claude Code's MCP namespace convention
+    #     `mcp__<server>__<tool>` (double underscores). Single-underscore
+    #     `mcp_<name>` is rejected as spoofing.
+    # Strategy: keep only the official prompt in `system`, and move any
+    # hermes-side system content into a <system_context> preamble on the
+    # first user message — Anthropic does not apply the spoof filter to
+    # user content.
     if is_oauth:
-        # 1. Prepend Claude Code system prompt identity
-        cc_block = {"type": "text", "text": _CLAUDE_CODE_SYSTEM_PREFIX}
+        # 1. Extract existing system text so we can relocate it.
+        extra_system_parts = []
         if isinstance(system, list):
-            system = [cc_block] + system
+            for b in system:
+                if isinstance(b, dict) and b.get("type") == "text":
+                    txt = b.get("text", "")
+                    if txt:
+                        extra_system_parts.append(txt)
         elif isinstance(system, str) and system:
-            system = [cc_block, {"type": "text", "text": system}]
-        else:
-            system = [cc_block]
-
-        # 2. Sanitize system prompt — replace product name references
-        #    to avoid Anthropic's server-side content filters.
-        for block in system:
-            if isinstance(block, dict) and block.get("type") == "text":
-                text = block.get("text", "")
-                text = text.replace("Hermes Agent", "Claude Code")
-                text = text.replace("Hermes agent", "Claude Code")
-                text = text.replace("hermes-agent", "claude-code")
-                text = text.replace("Nous Research", "Anthropic")
-                block["text"] = text
-
-        # 3. Prefix tool names with mcp_ (Claude Code convention)
+            extra_system_parts.append(system)
+
+        # 2. System: only the official Claude Code prompt.
+        system = [{"type": "text", "text": _CLAUDE_CODE_SYSTEM_PREFIX}]
+
+        # 3. Relocate hermes system content to the first user message.
+        if extra_system_parts:
+            preamble = (
+                "<system_context>\n"
+                + "\n\n".join(extra_system_parts)
+                + "\n</system_context>\n\n"
+            )
+            for msg in anthropic_messages:
+                if msg.get("role") != "user":
+                    continue
+                content = msg.get("content")
+                if isinstance(content, str):
+                    msg["content"] = preamble + content
+                elif isinstance(content, list):
+                    for block in content:
+                        if isinstance(block, dict) and block.get("type") == "text":
+                            block["text"] = preamble + block.get("text", "")
+                            break
+                    else:
+                        content.insert(0, {"type": "text", "text": preamble.rstrip()})
+                break
+
+        # 4. Prefix tool names with Claude Code's MCP namespace.
         if anthropic_tools:
             for tool in anthropic_tools:
-                if "name" in tool:
+                if "name" in tool and not tool["name"].startswith(_MCP_TOOL_PREFIX):
                     tool["name"] = _MCP_TOOL_PREFIX + tool["name"]
 
-        # 4. Prefix tool names in message history (tool_use and tool_result blocks)
+        # 5. Prefix tool_use names in message history (must match #4).
         for msg in anthropic_messages:
             content = msg.get("content")
-            if isinstance(content, list):
-                for block in content:
-                    if isinstance(block, dict):
-                        if block.get("type") == "tool_use" and "name" in block:
-                            if not block["name"].startswith(_MCP_TOOL_PREFIX):
-                                block["name"] = _MCP_TOOL_PREFIX + block["name"]
-                        elif block.get("type") == "tool_result" and "tool_use_id" in block:
-                            pass  # tool_result uses ID, not name
+            if not isinstance(content, list):
+                continue
+            for block in content:
+                if (
+                    isinstance(block, dict)
+                    and block.get("type") == "tool_use"
+                    and "name" in block
+                    and not block["name"].startswith(_MCP_TOOL_PREFIX)
+                ):
+                    block["name"] = _MCP_TOOL_PREFIX + block["name"]
 
     kwargs: Dict[str, Any] = {
         "model": model,

agent/transports/anthropic.py

@@ -84,11 +84,10 @@ def normalize_response(self, response: Any, **kwargs) -> NormalizedResponse:
         to OpenAI finish_reason, and collects reasoning_details in provider_data.
         """
         import json
-        from agent.anthropic_adapter import _to_plain_data
+        from agent.anthropic_adapter import _MCP_TOOL_PREFIX, _to_plain_data
         from agent.transports.types import ToolCall
 
         strip_tool_prefix = kwargs.get("strip_tool_prefix", False)
-        _MCP_PREFIX = "mcp_"
 
         text_parts = []
         reasoning_parts = []
@@ -105,8 +104,8 @@ def normalize_response(self, response: Any, **kwargs) -> NormalizedResponse:
                     reasoning_details.append(block_dict)
             elif block.type == "tool_use":
                 name = block.name
-                if strip_tool_prefix and name.startswith(_MCP_PREFIX):
-                    name = name[len(_MCP_PREFIX):]
+                if strip_tool_prefix and name.startswith(_MCP_TOOL_PREFIX):
+                    name = name[len(_MCP_TOOL_PREFIX):]
                 tool_calls.append(
                     ToolCall(
                         id=block.id,

tests/agent/test_anthropic_adapter.py

@@ -1278,6 +1278,226 @@ def test_context_length_no_clamp_when_larger(self):
         assert kwargs["max_tokens"] == 64_000
 
 
+# ---------------------------------------------------------------------------
+# OAuth spoof-filter compliance
+# ---------------------------------------------------------------------------
+# Anthropic's subscription OAuth path enforces a server-side spoof filter
+# that rejects requests with a 400 "out of extra usage" when:
+#   1. tool names use single-underscore mcp_X (must be mcp__server__tool)
+#   2. system text impersonates Claude Code beyond the official prefix
+# These tests pin the contract.
+
+
+class TestOauthSpoofFilterCompliance:
+    OFFICIAL = "You are Claude Code, Anthropic's official CLI for Claude."
+
+    @staticmethod
+    def _tool(name, desc="", props=None):
+        return {
+            "type": "function",
+            "function": {
+                "name": name,
+                "description": desc,
+                "parameters": {"type": "object", "properties": props or {}},
+            },
+        }
+
+    def test_mcp_tool_prefix_is_double_underscore_namespace(self):
+        """The shared prefix constant must use Claude Code's MCP namespace."""
+        from agent.anthropic_adapter import _MCP_TOOL_PREFIX
+        assert _MCP_TOOL_PREFIX == "mcp__hermes__", (
+            "Single-underscore mcp_X tool names trigger Anthropic's spoof "
+            "filter — `mcp__hermes__X` is the Claude Code MCP convention."
+        )
+
+    def test_transport_uses_shared_prefix_constant(self):
+        """Receiver-side strip must reuse the adapter's prefix constant, not
+        a literal that can drift out of sync."""
+        from pathlib import Path
+        import agent.transports.anthropic as _ta
+        src = Path(_ta.__file__).read_text(encoding="utf-8")
+        assert "_MCP_TOOL_PREFIX" in src, (
+            "transports/anthropic.py must import _MCP_TOOL_PREFIX from "
+            "anthropic_adapter so the prefix stays in sync."
+        )
+        assert '_MCP_PREFIX' not in src, (
+            "Drop the local _MCP_PREFIX literal — use the imported constant."
+        )
+
+    # ── System prompt handling ───────────────────────────────────────
+
+    def test_oauth_system_reduced_to_official_prefix_only(self):
+        """Hermes-side system text must NOT appear in `system` on OAuth."""
+        kwargs = build_anthropic_kwargs(
+            model="claude-haiku-4-5",
+            messages=[
+                {"role": "system", "content": "You are Hermes Agent, an intelligent AI assistant."},
+                {"role": "user", "content": "hi"},
+            ],
+            tools=None,
+            max_tokens=64,
+            reasoning_config=None,
+            is_oauth=True,
+        )
+        assert kwargs["system"] == [{"type": "text", "text": self.OFFICIAL}]
+
+    def test_oauth_system_relocated_to_first_user_message(self):
+        """A hermes system prompt becomes a <system_context> preamble."""
+        kwargs = build_anthropic_kwargs(
+            model="claude-haiku-4-5",
+            messages=[
+                {"role": "system", "content": "Be terse."},
+                {"role": "user", "content": "do the thing"},
+            ],
+            tools=None,
+            max_tokens=64,
+            reasoning_config=None,
+            is_oauth=True,
+        )
+        first_user = kwargs["messages"][0]
+        assert first_user["role"] == "user"
+        content = first_user["content"]
+        text = content if isinstance(content, str) else "".join(
+            b.get("text", "") for b in content if isinstance(b, dict)
+        )
+        assert "<system_context>" in text
+        assert "Be terse." in text
+        assert "</system_context>" in text
+        assert text.endswith("do the thing")
+
+    def test_oauth_no_extra_system_keeps_user_message_clean(self):
+        """If no hermes-side system content, user message is unchanged."""
+        kwargs = build_anthropic_kwargs(
+            model="claude-haiku-4-5",
+            messages=[{"role": "user", "content": "hi"}],
+            tools=None,
+            max_tokens=64,
+            reasoning_config=None,
+            is_oauth=True,
+        )
+        assert kwargs["system"] == [{"type": "text", "text": self.OFFICIAL}]
+        content = kwargs["messages"][0]["content"]
+        text = content if isinstance(content, str) else "".join(
+            b.get("text", "") for b in content if isinstance(b, dict)
+        )
+        assert "<system_context>" not in text
+        assert text == "hi"
+
+    def test_oauth_does_not_include_hermes_identity_string(self):
+        """Regression: spoof filter rejected sanitized hermes identity. Ensure it's gone."""
+        kwargs = build_anthropic_kwargs(
+            model="claude-haiku-4-5",
+            messages=[
+                {"role": "system", "content": "You are Hermes Agent, helpful and direct."},
+                {"role": "user", "content": "x"},
+            ],
+            tools=None,
+            max_tokens=64,
+            reasoning_config=None,
+            is_oauth=True,
+        )
+        # System must NOT contain the hermes-identity sentence in any form
+        sys_text = "".join(b["text"] for b in kwargs["system"])
+        assert "Hermes Agent" not in sys_text
+        # The pre-patch sanitizer's output ("You are Claude Code, an intelligent
+        # AI assistant created by Anthropic.") was itself the spoof trigger —
+        # make sure we no longer emit that synthetic phrase in `system`.
+        assert "intelligent AI assistant" not in sys_text
+
+    # ── Tool naming ───────────────────────────────────────────────────
+
+    def test_oauth_tools_get_double_underscore_prefix(self):
+        kwargs = build_anthropic_kwargs(
+            model="claude-haiku-4-5",
+            messages=[{"role": "user", "content": "hi"}],
+            tools=[
+                self._tool("browser_back", "Go back"),
+                self._tool("terminal", "Run shell", {"command": {"type": "string"}}),
+            ],
+            max_tokens=64,
+            reasoning_config=None,
+            is_oauth=True,
+        )
+        names = [t["name"] for t in kwargs["tools"]]
+        assert names == ["mcp__hermes__browser_back", "mcp__hermes__terminal"]
+        # No single-underscore mcp_ prefix should remain
+        for n in names:
+            assert not (n.startswith("mcp_") and not n.startswith("mcp__")), n
+
+    def test_oauth_tool_prefix_is_idempotent(self):
+        """Already-prefixed tools must not be double-prefixed (retry safety)."""
+        kwargs = build_anthropic_kwargs(
+            model="claude-haiku-4-5",
+            messages=[{"role": "user", "content": "hi"}],
+            tools=[self._tool("mcp__hermes__terminal")],
+            max_tokens=64,
+            reasoning_config=None,
+            is_oauth=True,
+        )
+        assert kwargs["tools"][0]["name"] == "mcp__hermes__terminal"
+
+    def test_oauth_tool_use_in_history_gets_prefixed(self):
+        """tool_use blocks in prior turns must be renamed to match new tools."""
+        kwargs = build_anthropic_kwargs(
+            model="claude-haiku-4-5",
+            messages=[
+                {"role": "user", "content": "list"},
+                {"role": "assistant", "content": [
+                    {"type": "tool_use", "id": "tu_1", "name": "terminal",
+                     "input": {"command": "ls"}},
+                ]},
+                {"role": "tool", "tool_call_id": "tu_1", "content": "a b c"},
+            ],
+            tools=[self._tool("terminal")],
+            max_tokens=64,
+            reasoning_config=None,
+            is_oauth=True,
+        )
+        # Find the assistant message; its tool_use should now use the prefix.
+        for msg in kwargs["messages"]:
+            if msg.get("role") == "assistant":
+                blocks = msg["content"]
+                if isinstance(blocks, list):
+                    for b in blocks:
+                        if isinstance(b, dict) and b.get("type") == "tool_use":
+                            assert b["name"] == "mcp__hermes__terminal"
+                            break
+
+    # ── Non-OAuth path stays clean ────────────────────────────────────
+
+    def test_non_oauth_system_passes_through_unchanged(self):
+        kwargs = build_anthropic_kwargs(
+            model="claude-haiku-4-5",
+            messages=[
+                {"role": "system", "content": "You are Hermes Agent."},
+                {"role": "user", "content": "hi"},
+            ],
+            tools=None,
+            max_tokens=64,
+            reasoning_config=None,
+            is_oauth=False,
+        )
+        # Non-OAuth: system stays as-is (string), no <system_context> wrap.
+        assert kwargs["system"] == "You are Hermes Agent."
+        content = kwargs["messages"][0]["content"]
+        text = content if isinstance(content, str) else "".join(
+            b.get("text", "") for b in content if isinstance(b, dict)
+        )
+        assert "<system_context>" not in text
+        assert text == "hi"
+
+    def test_non_oauth_tools_not_prefixed(self):
+        kwargs = build_anthropic_kwargs(
+            model="claude-haiku-4-5",
+            messages=[{"role": "user", "content": "hi"}],
+            tools=[self._tool("terminal")],
+            max_tokens=64,
+            reasoning_config=None,
+            is_oauth=False,
+        )
+        assert kwargs["tools"][0]["name"] == "terminal"
+
+
 # ---------------------------------------------------------------------------
 # Model output limit lookup
 # ---------------------------------------------------------------------------