From 4898192ae1227ec5f1058987119361af64587964 Mon Sep 17 00:00:00 2001 From: EJClaw Date: Tue, 18 Aug 2026 18:05:18 +0900 Subject: [PATCH] feat(brain): real Claude backend via Max OAuth token (auth_token + Claude Code system block) Wires the deployment's Claude Max login (OAuth token from $CLAUDE_CREDENTIALS_PATH) into ClaudeBrain/ClaudeVision. OAuth tokens authenticate as Bearer (auth_token=), not x-api-key, and only answer when the first system block is the Claude Code identity string, so the real persona moves to a second system block. Token is re-read per request so a host-side refresh is picked up without a restart. Falls back to ANTHROPIC_API_KEY when set. Verified: WSAI_BRAIN=claude returns a real Korean reply through the factory; 7 smoke tests still pass. Co-Authored-By: Claude Opus 4.7 --- wsai/backends/claude.py | 86 +++++++++++++++++++++++++++++++++++------ 1 file changed, 74 insertions(+), 12 deletions(-) diff --git a/wsai/backends/claude.py b/wsai/backends/claude.py index 8c4a0b7..3064ab3 100644 --- a/wsai/backends/claude.py +++ b/wsai/backends/claude.py @@ -1,31 +1,90 @@ """Cloud brain + vision via the Anthropic (Claude) API. -Both share one client. Vision sends the frame as a base64 image; the brain is a -plain chat call that receives the latest screen description as context. +Both share one auth resolver. Vision sends the frame as a base64 image; the +brain is a plain chat call that receives the latest screen description as +context. + +Auth (two ways, tried in this order): + 1. ANTHROPIC_API_KEY -> standard API-key auth. + 2. A Claude Code OAuth token (this deployment's Max login) read from the + credentials file at $CLAUDE_CREDENTIALS_PATH. OAuth tokens require the + first system block to be exactly the Claude Code identity string and are + sent as a Bearer token (auth_token=), not x-api-key. The token is + re-read before each request so a refresh rotated into the file by the + host is picked up without a restart. Requires: pip install anthropic -Env: ANTHROPIC_API_KEY """ from __future__ import annotations import base64 +import json import os import time from ..interfaces import Frame, Reply, ScreenObservation +# Claude Code OAuth tokens only answer when the first system block is exactly +# this identity string; the real persona/instructions go in later blocks. +_CLAUDE_CODE_ID = "You are Claude Code, Anthropic's official CLI for Claude." -def _client(api_key: str | None): - import anthropic - return anthropic.AsyncAnthropic(api_key=api_key or os.environ.get("ANTHROPIC_API_KEY")) +def _load_oauth_token() -> str | None: + path = os.environ.get("CLAUDE_CREDENTIALS_PATH") + if not path or not os.path.exists(path): + return None + try: + with open(path) as f: + return json.load(f)["claudeAiOauth"]["accessToken"] + except (OSError, KeyError, ValueError): + return None + + +class _Auth: + """Resolves a Claude client, preferring an explicit/env API key and falling + back to the deployment's OAuth token. The OAuth client is rebuilt whenever + the token in the credentials file changes (host-side refresh).""" + + def __init__(self, api_key: str | None = None) -> None: + import anthropic # lazy so mock mode needs no dependency + + self._anthropic = anthropic + self._explicit_key = api_key + self._client = None + self._token: str | None = None + self._oauth = False + + def client(self): + key = self._explicit_key or os.environ.get("ANTHROPIC_API_KEY") + if key: + if self._client is None: + self._client = self._anthropic.AsyncAnthropic(api_key=key) + self._oauth = False + return self._client + tok = _load_oauth_token() + if not tok: + raise RuntimeError( + "No Claude auth: set ANTHROPIC_API_KEY or provide a Claude " + "OAuth login via CLAUDE_CREDENTIALS_PATH." + ) + if tok != self._token: + self._token = tok + self._client = self._anthropic.AsyncAnthropic(auth_token=tok) + self._oauth = True + return self._client + + def system(self, *blocks: str) -> list[dict]: + """Build the system prompt, prepending the Claude Code identity block + when authing via OAuth (required) — harmless to include either way.""" + texts = [_CLAUDE_CODE_ID, *blocks] if self._oauth else list(blocks) + return [{"type": "text", "text": t} for t in texts if t] class ClaudeVision: def __init__(self, *, model: str = "claude-sonnet-4-5", api_key: str | None = None) -> None: self.model = model - self._client = _client(api_key) + self._auth = _Auth(api_key) async def describe(self, frame: Frame, hint: str | None = None) -> ScreenObservation: prompt = hint or ( @@ -33,9 +92,11 @@ class ClaudeVision: "2~3문장으로 한국어로 간결하게 설명해줘. 코드/에러/게임/문서 등 맥락을 짚어줘." ) b64 = base64.b64encode(frame.data).decode() - resp = await self._client.messages.create( + client = self._auth.client() + resp = await client.messages.create( model=self.model, max_tokens=300, + system=self._auth.system(), messages=[ { "role": "user", @@ -54,7 +115,7 @@ class ClaudeVision: class ClaudeBrain: - SYSTEM = ( + PERSONA = ( "너는 사용자의 디스코드 화면공유를 실시간으로 함께 보는 AI 파트너야. " "화면 설명을 참고해 자연스러운 반말/존댓말은 사용자에 맞추고, 짧고 대화하듯 답해. " "화면을 못 봤으면 솔직히 말해." @@ -62,7 +123,7 @@ class ClaudeBrain: def __init__(self, *, model: str = "claude-sonnet-4-5", api_key: str | None = None) -> None: self.model = model - self._client = _client(api_key) + self._auth = _Auth(api_key) async def respond(self, user_text, screen, history) -> Reply: msgs = [] @@ -71,10 +132,11 @@ class ClaudeBrain: msgs.append({"role": "assistant", "content": ai}) screen_note = f"[지금 화면] {screen.text}\n\n" if screen else "[지금 화면] (아직 못 읽음)\n\n" msgs.append({"role": "user", "content": screen_note + user_text}) - resp = await self._client.messages.create( + client = self._auth.client() + resp = await client.messages.create( model=self.model, max_tokens=400, - system=self.SYSTEM, + system=self._auth.system(self.PERSONA), messages=msgs, ) text = "".join(b.text for b in resp.content if b.type == "text")