diff --git a/README.md b/README.md index bc168a4..d3cae32 100644 --- a/README.md +++ b/README.md @@ -1,5 +1,75 @@ # Amplifier Anthropic Provider Module +This is a drop-in fork of +[`microsoft/amplifier-module-provider-anthropic`](https://github.com/microsoft/amplifier-module-provider-anthropic). +It keeps the official provider implementation and adds Claude Pro/Max OAuth +using the same direct Anthropic Messages API approach as Pi. + +## Quickstart + +Copy and paste: + +```bash +curl -LsSf https://astral.sh/uv/install.sh | sh +uv tool install --force git+https://github.com/microsoft/amplifier + +FORK=git+https://github.com/gszep/amplifier-module-provider-anthropic@main +amplifier source add provider-anthropic "$FORK" --global --module +amplifier provider install anthropic --force +uvx --refresh --from "$FORK" amplifier-anthropic-login +amplifier provider add anthropic +amplifier provider test anthropic +``` + +Leave the API-key prompt blank after OAuth login, then run `amplifier`. + +OAuth credentials are stored in `~/.amplifier/anthropic-auth.json` with mode +`0600` and refreshed automatically. Authentication precedence is +`ANTHROPIC_OAUTH_TOKEN`, stored OAuth, configured `api_key`, then +`ANTHROPIC_API_KEY`. + +OAuth requests use bearer authentication, Claude Code identity headers and +system identity, and canonical Claude Code casing for matching built-in tool +names. Tools otherwise follow the official provider's native `tools`, +`tool_use`, and `tool_result` path; nothing is serialized into model-visible +text. The request contract is centralized in +`amplifier_anthropic_oauth/auth.py` and checked against an installed Claude Code +executable by `tests/test_claude_header_parity.py`. + +### OAuth test coverage + +The automated suite includes transport-level assertions over the final HTTP +requests emitted by the Anthropic SDK for both `/v1/models` and `/v1/messages`. +It verifies bearer auth, absence of `x-api-key`, Claude Code user-agent and +`x-app`, and the complete required beta-header set. Token exchange/refresh +request construction is tested separately, including its OAuth identity +headers and error bodies. + +The live OAuth test is excluded from normal and CI runs. It uses the stored +provider credential to list models and force a native tool call through the +Messages API: + +```bash +uv run pytest -m live_oauth +``` + +Normal local test runs also execute the header-capture test whenever `claude` +and `~/.claude/.credentials.json` are available; CI runners skip it. The test +starts a minimal in-process CONNECT proxy, +generates a temporary CA and leaf certificate, passes that CA to Claude through +`NODE_EXTRA_CA_CERTS`, captures one `claude -p` Messages request, immediately +redacts its bearer token, compares stable OAuth headers, and tears everything +down: + +```bash +uv run pytest -m local_header_capture +``` + +No container, persistent CA installation, or third-party proxy is required. +`tests/test_claude_header_parity.py` separately checks the client ID, OAuth +endpoints, betas, and user-agent version embedded in the installed Claude Code +executable. + Claude model integration for Amplifier via Anthropic API. ## Prerequisites diff --git a/amplifier_anthropic_oauth/__init__.py b/amplifier_anthropic_oauth/__init__.py new file mode 100644 index 0000000..1b06a1f --- /dev/null +++ b/amplifier_anthropic_oauth/__init__.py @@ -0,0 +1 @@ +"""Standalone Claude Pro/Max OAuth support for the Amplifier Anthropic provider.""" diff --git a/amplifier_anthropic_oauth/auth.py b/amplifier_anthropic_oauth/auth.py new file mode 100644 index 0000000..296ae58 --- /dev/null +++ b/amplifier_anthropic_oauth/auth.py @@ -0,0 +1,295 @@ +"""Anthropic OAuth credential management for Claude Pro/Max accounts.""" + +from __future__ import annotations + +import asyncio +import base64 +from dataclasses import dataclass +import json +import os +from pathlib import Path +import secrets +import subprocess +import time +from typing import Any +from urllib.parse import parse_qs, urlencode, urlparse +from urllib.error import HTTPError +from urllib.request import Request, urlopen + + +CLIENT_ID = base64.b64decode( + "OWQxYzI1MGEtZTYxYi00NGQ5LTg4ZWQtNTk0NGQxOTYyZjVl" +).decode() +AUTHORIZE_URL = "https://claude.com/cai/oauth/authorize" +TOKEN_URL = "https://platform.claude.com/v1/oauth/token" +CALLBACK_HOST = "127.0.0.1" +CALLBACK_PORT = 53692 +CALLBACK_PATH = "/callback" +REDIRECT_URI = f"http://localhost:{CALLBACK_PORT}{CALLBACK_PATH}" +SCOPES = ( + "org:create_api_key user:profile user:inference " + "user:sessions:claude_code user:mcp_servers user:file_upload" +) +# Stable OAuth identity betas. Feature-specific betas (thinking, context, +# tools, caching) are selected by the upstream provider per request. +OAUTH_BETAS = ( + "claude-code-20250219", + "oauth-2025-04-20", +) + + +def installed_claude_code_version() -> str: + """Use the installed CLI version in the identity header when available.""" + configured = os.environ.get("AMPLIFIER_CLAUDE_CODE_VERSION") + if configured: + return configured + try: + result = subprocess.run( + ["claude", "--version"], + capture_output=True, + text=True, + timeout=2, + check=False, + ) + version = result.stdout.strip().split(" ", 1)[0] + if version and all(part.isdigit() for part in version.split(".")): + return version + except (OSError, subprocess.SubprocessError): + pass + # This is only an attribution header. Authentication does not depend on + # the installed CLI, so retain a known-compatible fallback. + return "2.1.75" + + +def oauth_request_headers() -> dict[str, str]: + """Headers used by pi for Anthropic Claude Pro/Max OAuth requests.""" + return { + "Accept": "application/json", + "anthropic-dangerous-direct-browser-access": "true", + "anthropic-beta": ",".join(OAUTH_BETAS), + # Match the SDK's canonical key casing so this replaces, rather than + # appends to, its default AsyncAnthropic/Python user-agent. + "User-Agent": ( + f"claude-cli/{installed_claude_code_version()} (external, sdk-cli)" + ), + "x-app": "cli", + } + + +@dataclass(frozen=True) +class AnthropicAuth: + """Resolved request authentication.""" + + token: str + oauth: bool + + +class AnthropicAuthError(RuntimeError): + """Raised when Anthropic credentials cannot be resolved or refreshed.""" + + +def default_auth_path() -> Path: + configured = os.environ.get("AMPLIFIER_ANTHROPIC_AUTH_FILE") or os.environ.get( + "AMPLIFIER_CLAUDE_AUTH_FILE" + ) + return ( + Path(configured).expanduser() + if configured + else Path.home() / ".amplifier" / "anthropic-auth.json" + ) + + +def _b64url(value: bytes) -> str: + return base64.urlsafe_b64encode(value).decode().rstrip("=") + + +def generate_pkce() -> tuple[str, str]: + import hashlib + + verifier = _b64url(secrets.token_bytes(32)) + challenge = _b64url(hashlib.sha256(verifier.encode()).digest()) + return verifier, challenge + + +def authorization_url(verifier: str, challenge: str) -> str: + query = urlencode( + { + "code": "true", + "client_id": CLIENT_ID, + "response_type": "code", + "redirect_uri": REDIRECT_URI, + "scope": SCOPES, + "code_challenge": challenge, + "code_challenge_method": "S256", + "state": verifier, + } + ) + return f"{AUTHORIZE_URL}?{query}" + + +def parse_authorization_input(value: str) -> tuple[str | None, str | None]: + value = value.strip() + if not value: + return None, None + try: + parsed = urlparse(value) + if parsed.scheme and parsed.netloc: + query = parse_qs(parsed.query) + return query.get("code", [None])[0], query.get("state", [None])[0] + except ValueError: + pass + if "#" in value: + code, state = value.split("#", 1) + return code or None, state or None + if "code=" in value: + query = parse_qs(value) + return query.get("code", [None])[0], query.get("state", [None])[0] + return value, None + + +def _post_json(url: str, body: dict[str, Any]) -> dict[str, Any]: + identity_headers = oauth_request_headers() + request = Request( + url, + data=json.dumps(body).encode(), + headers={ + "Content-Type": "application/json", + "Accept": "application/json", + "User-Agent": identity_headers["User-Agent"], + "anthropic-beta": "oauth-2025-04-20", + "x-app": identity_headers["x-app"], + }, + method="POST", + ) + try: + with urlopen(request, timeout=30) as response: # noqa: S310 - fixed HTTPS endpoint + payload = response.read().decode() + except HTTPError as exc: + response_body = exc.read().decode(errors="replace") + raise AnthropicAuthError( + f"Anthropic OAuth request failed: HTTP {exc.code}: {response_body}" + ) from exc + except Exception as exc: + raise AnthropicAuthError(f"Anthropic OAuth request failed: {exc}") from exc + try: + result = json.loads(payload) + except json.JSONDecodeError as exc: + raise AnthropicAuthError("Anthropic OAuth returned invalid JSON") from exc + if not isinstance(result, dict): + raise AnthropicAuthError("Anthropic OAuth returned an invalid response") + return result + + +def _credentials_from_token_response(data: dict[str, Any]) -> dict[str, Any]: + access = data.get("access_token") + refresh = data.get("refresh_token") + expires_in = data.get("expires_in") + if not isinstance(access, str) or not isinstance(refresh, str): + raise AnthropicAuthError("Anthropic OAuth response did not contain tokens") + if not isinstance(expires_in, (int, float)): + raise AnthropicAuthError("Anthropic OAuth response did not contain an expiry") + return { + "type": "oauth", + "access": access, + "refresh": refresh, + # Refresh five minutes early, matching pi's credential handling. + "expires": int(time.time() * 1000 + expires_in * 1000 - 5 * 60 * 1000), + } + + +def exchange_authorization_code(code: str, state: str, verifier: str) -> dict[str, Any]: + return _credentials_from_token_response( + _post_json( + TOKEN_URL, + { + "grant_type": "authorization_code", + "client_id": CLIENT_ID, + "code": code, + "state": state, + "redirect_uri": REDIRECT_URI, + "code_verifier": verifier, + }, + ) + ) + + +def refresh_oauth_credentials(credentials: dict[str, Any]) -> dict[str, Any]: + refresh = credentials.get("refresh") + if not isinstance(refresh, str) or not refresh: + raise AnthropicAuthError("Stored Anthropic OAuth credentials have no refresh token") + return _credentials_from_token_response( + _post_json( + TOKEN_URL, + { + "grant_type": "refresh_token", + "client_id": CLIENT_ID, + "refresh_token": refresh, + "scope": SCOPES, + }, + ) + ) + + +def read_credentials(path: Path) -> dict[str, Any] | None: + try: + value = json.loads(path.read_text()) + except FileNotFoundError: + return None + except (OSError, json.JSONDecodeError) as exc: + raise AnthropicAuthError(f"Could not read Anthropic credentials at {path}: {exc}") from exc + if not isinstance(value, dict): + raise AnthropicAuthError(f"Invalid Anthropic credentials at {path}") + return value + + +def write_credentials(path: Path, credentials: dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + temporary = path.with_name(f".{path.name}.{os.getpid()}.{secrets.token_hex(4)}.tmp") + try: + fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + with os.fdopen(fd, "w", encoding="utf-8") as stream: + json.dump(credentials, stream, indent=2) + stream.write("\n") + stream.flush() + os.fsync(stream.fileno()) + os.replace(temporary, path) + os.chmod(path, 0o600) + finally: + try: + temporary.unlink() + except FileNotFoundError: + pass + + +class AnthropicAuthManager: + """Resolve API-key or OAuth auth and refresh OAuth tokens when needed.""" + + def __init__(self, path: Path | None = None, api_key: str | None = None) -> None: + self.path = path or default_auth_path() + self.api_key = api_key + self._lock = asyncio.Lock() + + async def get_auth(self) -> AnthropicAuth: + oauth_token = os.environ.get("ANTHROPIC_OAUTH_TOKEN") + if oauth_token: + return AnthropicAuth(oauth_token, oauth=True) + + async with self._lock: + credentials = await asyncio.to_thread(read_credentials, self.path) + if credentials and credentials.get("type") == "oauth": + expires = credentials.get("expires", 0) + if not isinstance(expires, (int, float)) or expires <= time.time() * 1000: + credentials = await asyncio.to_thread(refresh_oauth_credentials, credentials) + await asyncio.to_thread(write_credentials, self.path, credentials) + access = credentials.get("access") + if isinstance(access, str) and access: + return AnthropicAuth(access, oauth=True) + + key = self.api_key or os.environ.get("ANTHROPIC_API_KEY") + if key: + return AnthropicAuth(key, oauth=False) + + raise AnthropicAuthError( + "No Anthropic credentials. Run `amplifier-anthropic-login`, set " + "ANTHROPIC_OAUTH_TOKEN`, or set `ANTHROPIC_API_KEY`." + ) diff --git a/amplifier_anthropic_oauth/login.py b/amplifier_anthropic_oauth/login.py new file mode 100644 index 0000000..7e1dac7 --- /dev/null +++ b/amplifier_anthropic_oauth/login.py @@ -0,0 +1,186 @@ +"""Command-line login for Anthropic Claude Pro/Max OAuth.""" + +from __future__ import annotations + +import asyncio +from contextlib import suppress +import os +import sys +import webbrowser + +from .auth import ( + AnthropicAuthError, + CALLBACK_HOST, + CALLBACK_PATH, + CALLBACK_PORT, + REDIRECT_URI, + authorization_url, + default_auth_path, + exchange_authorization_code, + generate_pkce, + parse_authorization_input, + write_credentials, +) + +_SUCCESS = """Anthropic login complete +

Authentication complete

You can close this window.

""" +_ERROR = """Anthropic login failed +

Authentication failed

{message}

""" + + +async def _read_callback( + reader: asyncio.StreamReader, + writer: asyncio.StreamWriter, + expected_state: str, + result: asyncio.Future[tuple[str, str]], +) -> None: + status = "400 Bad Request" + body = _ERROR.format(message="Missing authorization response.") + try: + request_line = (await reader.readline()).decode(errors="replace").strip() + parts = request_line.split(" ") + if len(parts) >= 2: + code, state = parse_authorization_input(f"http://localhost{parts[1]}") + if not parts[1].startswith(CALLBACK_PATH): + status = "404 Not Found" + body = _ERROR.format(message="Callback route not found.") + elif not code or not state: + body = _ERROR.format(message="Missing code or state parameter.") + elif state != expected_state: + body = _ERROR.format(message="OAuth state mismatch.") + else: + status = "200 OK" + body = _SUCCESS + if not result.done(): + result.set_result((code, state)) + except Exception as exc: + body = _ERROR.format(message=str(exc)) + payload = body.encode() + writer.write( + f"HTTP/1.1 {status}\r\nContent-Type: text/html; charset=utf-8\r\n" + f"Content-Length: {len(payload)}\r\nConnection: close\r\n\r\n".encode() + + payload + ) + with suppress(Exception): + await writer.drain() + writer.close() + with suppress(Exception): + await writer.wait_closed() + + +async def login() -> None: + verifier, challenge = generate_pkce() + url = authorization_url(verifier, challenge) + loop = asyncio.get_running_loop() + callback: asyncio.Future[tuple[str, str]] = loop.create_future() + + server: asyncio.Server | None = None + try: + server = await asyncio.start_server( + lambda reader, writer: _read_callback(reader, writer, verifier, callback), + CALLBACK_HOST, + CALLBACK_PORT, + ) + except OSError as exc: + print(f"Could not start the local callback server: {exc}", file=sys.stderr) + + print("Open this URL to authenticate with Anthropic:\n") + print(url) + print(f"\nWaiting for the callback at {REDIRECT_URI}.") + opened = webbrowser.open(url) + if not opened: + print("Could not open a browser automatically; open the URL above manually.") + + manual: asyncio.Future[str] = loop.create_future() + manual_input = bytearray() + stdin_fd: int | None = None + stdin_was_blocking: bool | None = None + + def read_manual_input() -> None: + # An asyncio fd callback must never call TextIO.readline(): terminals + # can report a partial/spurious readiness event, after which readline + # blocks the entire event loop even though the browser callback won. + assert stdin_fd is not None + try: + chunk = os.read(stdin_fd, 4096) + except BlockingIOError: + return + if chunk: + manual_input.extend(chunk) + if not chunk or b"\n" in manual_input or b"\r" in manual_input: + line = bytes(manual_input).splitlines()[0] if manual_input else b"" + if not manual.done(): + manual.set_result(line.decode(errors="replace")) + + has_stdin_reader = False + try: + stdin_fd = sys.stdin.fileno() + stdin_was_blocking = os.get_blocking(stdin_fd) + os.set_blocking(stdin_fd, False) + loop.add_reader(stdin_fd, read_manual_input) + has_stdin_reader = True + print( + "No terminal input is needed when the browser is on this machine.\n" + "If the browser is elsewhere, paste the final redirect URL or " + "authorization code here, then press Enter:\n> ", + end="", + flush=True, + ) + except (AttributeError, NotImplementedError, OSError): + if stdin_fd is not None and stdin_was_blocking is not None: + with suppress(OSError): + os.set_blocking(stdin_fd, stdin_was_blocking) + stdin_fd = None + + waiters = {callback, manual} if has_stdin_reader else {callback} + try: + try: + async with asyncio.timeout(5 * 60): + done, _ = await asyncio.wait( + waiters, return_when=asyncio.FIRST_COMPLETED + ) + except TimeoutError as exc: + raise AnthropicAuthError( + "Timed out waiting for the OAuth callback" + ) from exc + if callback in done: + code, state = callback.result() + print("\nOAuth callback received; exchanging authorization code...") + else: + code, supplied_state = parse_authorization_input(manual.result()) + state = supplied_state or verifier + if supplied_state and supplied_state != verifier: + raise AnthropicAuthError("OAuth state mismatch") + if not code: + raise AnthropicAuthError("Missing authorization code") + + credentials = await asyncio.to_thread( + exchange_authorization_code, code, state, verifier + ) + path = default_auth_path() + await asyncio.to_thread(write_credentials, path, credentials) + print(f"Anthropic OAuth credentials saved to {path}") + finally: + if has_stdin_reader and stdin_fd is not None: + loop.remove_reader(stdin_fd) + if stdin_fd is not None and stdin_was_blocking is not None: + with suppress(OSError): + os.set_blocking(stdin_fd, stdin_was_blocking) + if server: + server.close() + await server.wait_closed() + for task in (manual, callback): + if not task.done(): + task.cancel() + + +def main() -> None: + try: + asyncio.run(login()) + except (AnthropicAuthError, KeyboardInterrupt) as exc: + print(f"Login failed: {exc}", file=sys.stderr) + raise SystemExit(1) from exc + + +if __name__ == "__main__": + main() diff --git a/amplifier_module_provider_anthropic/__init__.py b/amplifier_module_provider_anthropic/__init__.py index 6bec4a0..2df355f 100644 --- a/amplifier_module_provider_anthropic/__init__.py +++ b/amplifier_module_provider_anthropic/__init__.py @@ -17,6 +17,7 @@ import time import uuid from decimal import Decimal +from pathlib import Path from threading import Lock from typing import Any @@ -59,6 +60,13 @@ ) # Not exported in public API as of SDK v0.96.0 (private import still works) from ._cost import compute_cost +from amplifier_anthropic_oauth.auth import ( + AnthropicAuth, + AnthropicAuthError, + AnthropicAuthManager, + OAUTH_BETAS, + oauth_request_headers, +) @dataclass @@ -228,6 +236,28 @@ async def _get_process_semaphore(max_concurrent: int) -> asyncio.Semaphore | Non BETA_HEADER_INTERLEAVED_THINKING = "interleaved-thinking-2025-05-14" BETA_HEADER_TASK_BUDGETS = "task-budgets-2026-03-13" BETA_HEADER_FAST_MODE = "fast-mode-2026-02-01" +_CLAUDE_CODE_TOOL_NAMES = { + name.lower(): name + for name in ( + "Read", + "Write", + "Edit", + "Bash", + "Grep", + "Glob", + "AskUserQuestion", + "EnterPlanMode", + "ExitPlanMode", + "KillShell", + "NotebookEdit", + "Skill", + "Task", + "TaskOutput", + "TodoWrite", + "WebFetch", + "WebSearch", + ) +} PROVIDER_FALLBACK_OPEN = "provider:fallback_open" PROVIDER_FALLBACK_ACTIVE = "provider:fallback_active" FALLBACK_STATE_VERSION = 1 @@ -381,16 +411,25 @@ def _add_cost(cost) -> None: _totals["cost_usd"] = (_totals["cost_usd"] or Decimal("0")) + cost _totals["has_data"] = True - # Get API key from config or environment - api_key = config.get("api_key") - if not api_key: - api_key = os.environ.get("ANTHROPIC_API_KEY") - - if not api_key: - logger.warning("No API key found for Anthropic provider") + auth_path = config.get("auth_file") + auth_manager = AnthropicAuthManager( + path=Path(auth_path).expanduser() if auth_path else None, + api_key=config.get("api_key"), + ) + try: + initial_auth = await auth_manager.get_auth() + except AnthropicAuthError as exc: + logger.warning("Anthropic authentication unavailable: %s", exc) return None - provider = AnthropicProvider(api_key, config, coordinator, add_cost=_add_cost) + provider = AnthropicProvider( + initial_auth.token, + config, + coordinator, + add_cost=_add_cost, + auth_manager=auth_manager, + initial_auth=initial_auth, + ) await coordinator.mount("providers", provider, name="anthropic") coordinator.register_contributor( "session.cost", @@ -473,6 +512,8 @@ def __init__( config: dict[str, Any] | None = None, coordinator: ModuleCoordinator | None = None, add_cost=None, + auth_manager: AnthropicAuthManager | None = None, + initial_auth: AnthropicAuth | None = None, ): """ Initialize Anthropic provider. @@ -485,9 +526,25 @@ def __init__( config: Additional configuration coordinator: Module coordinator for event emission """ + # Amplifier's configuration-time loader passes an empty string when an + # optional API-key field is left blank. Treat it as absent so stored + # OAuth credentials remain discoverable during model listing/testing. + api_key = api_key or None self._api_key = api_key - self._client: AsyncAnthropic | None = None # Lazy init self.config = config or {} + auth_path = self.config.get("auth_file") + self._auth_manager = auth_manager or ( + AnthropicAuthManager( + path=Path(auth_path).expanduser() if auth_path else None, + api_key=None, + ) + if api_key is None + else None + ) + self._auth_state = initial_auth or ( + AnthropicAuth(api_key, oauth=False) if api_key else None + ) + self._client: AsyncAnthropic | None = None # Lazy init self.coordinator = coordinator self.default_model = self.config.get("default_model", "claude-sonnet-4-5") self._default_caps = self._get_capabilities(self.default_model) @@ -628,6 +685,15 @@ def __init__( self._default_headers = {"anthropic-beta": beta_header_value} logger.info(f"[PROVIDER] Beta headers enabled: {beta_header_value}") + if self._auth_state and self._auth_state.oauth: + self._beta_headers = list( + dict.fromkeys([*OAUTH_BETAS, *self._beta_headers]) + ) + self._default_headers = { + **oauth_request_headers(), + "anthropic-beta": ",".join(self._beta_headers), + } + # Shared rate-limit state file for cross-process awareness. # All Anthropic provider instances (across processes, Docker containers # sharing a filesystem, etc.) read this file before the per-emptive @@ -670,6 +736,7 @@ def __init__( # detected repeatedly across LLM iterations (since synthetic results # are injected into request.messages but not persisted to message store). self._repaired_tool_ids: set[str] = set() + self._oauth_tool_names: dict[str, str] = {} self._add_cost = add_cost or (lambda cost: None) @property @@ -680,20 +747,57 @@ def client(self) -> AsyncAnthropic: raise ValueError("api_key must be provided for API calls") # Set SDK max_retries=0 - we handle retries ourselves to properly # honor retry-after headers with jitter and longer backoffs - self._client = AsyncAnthropic( - api_key=self._api_key, - base_url=self._base_url, - default_headers=self._default_headers, - max_retries=0, - ) + auth = self._auth_state + if auth and auth.oauth: + # Passing api_key=None makes the SDK silently reload + # ANTHROPIC_API_KEY from the environment. Use an explicit empty + # value during construction, then clear it so OAuth requests + # carry Authorization only and never an X-Api-Key header. + self._client = AsyncAnthropic( + api_key="", + auth_token=auth.token, + base_url=self._base_url, + default_headers=self._default_headers, + max_retries=0, + ) + self._client.api_key = None + else: + self._client = AsyncAnthropic( + api_key=self._api_key, + base_url=self._base_url, + default_headers=self._default_headers, + max_retries=0, + ) return self._client + async def _refresh_auth(self) -> None: + """Refresh OAuth credentials and rotate the SDK client when needed.""" + if self._auth_manager is None: + return + auth = await self._auth_manager.get_auth() + if auth == self._auth_state: + return + old_client = self._client + self._client = None + self._auth_state = auth + self._api_key = auth.token + if auth.oauth: + self._beta_headers = list( + dict.fromkeys([*OAUTH_BETAS, *self._beta_headers]) + ) + self._default_headers = { + **oauth_request_headers(), + "anthropic-beta": ",".join(self._beta_headers), + } + if old_client is not None: + await old_client.close() + def get_info(self) -> ProviderInfo: """Get provider metadata.""" return ProviderInfo( id="anthropic", display_name="Anthropic", - credential_env_vars=["ANTHROPIC_API_KEY"], + credential_env_vars=["ANTHROPIC_OAUTH_TOKEN", "ANTHROPIC_API_KEY"], capabilities=list(self._default_caps.capability_tags), defaults={ "model": self.default_model, @@ -708,10 +812,11 @@ def get_info(self) -> ProviderInfo: config_fields=[ ConfigField( id="api_key", - display_name="API Key", + display_name="API Key (optional with Claude Pro/Max OAuth)", field_type="secret", - prompt="Enter your Anthropic API key", + prompt="Enter your Anthropic API key, or leave blank after OAuth login", env_var="ANTHROPIC_API_KEY", + required=False, ), ConfigField( id="base_url", @@ -852,6 +957,7 @@ async def list_models(self) -> list[ModelInfo]: Returns: List of ModelInfo for available Claude models. """ + await self._refresh_auth() response = await self.client.models.list() api_models = list(response.data) @@ -1851,6 +1957,8 @@ async def complete(self, request: ChatRequest, **kwargs) -> ChatResponse: Returns: ChatResponse with content blocks, tool calls, usage """ + await self._refresh_auth() + # VALIDATE AND REPAIR: Check for missing tool results (backup safety net) missing = self._find_missing_tool_results(request.messages) @@ -2160,6 +2268,38 @@ def _format_system_with_cache( return [block] + def _apply_oauth_request_contract(self, params: dict[str, Any]) -> None: + """Apply Claude Code identity and canonical tool casing for OAuth.""" + if not self._auth_state or not self._auth_state.oauth: + return + + self._oauth_tool_names = {} + for tool in params.get("tools", []): + name = tool.get("name") + if isinstance(name, str): + canonical = _CLAUDE_CODE_TOOL_NAMES.get(name.lower(), name) + self._oauth_tool_names[canonical.lower()] = name + tool["name"] = canonical + + for message in params.get("messages", []): + content = message.get("content") + if not isinstance(content, list): + continue + for block in content: + if block.get("type") != "tool_use": + continue + name = block.get("name") + if isinstance(name, str): + block["name"] = _CLAUDE_CODE_TOOL_NAMES.get(name.lower(), name) + + identity: dict[str, Any] = { + "type": "text", + "text": "You are Claude Code, Anthropic's official CLI for Claude.", + } + if self.enable_prompt_caching: + identity["cache_control"] = {"type": "ephemeral"} + params["system"] = [identity, *(params.get("system") or [])] + async def _complete_chat_request( self, request: ChatRequest, @@ -2610,8 +2750,10 @@ async def _complete_chat_request( extra_headers["anthropic-beta"] = ",".join(request_beta_headers) params["extra_headers"] = extra_headers + self._apply_oauth_request_contract(params) + logger.info( - f"[PROVIDER] Anthropic API call - model: {params['model']}, messages: {len(params['messages'])}, system: {bool(system_blocks)}, tools: {len(params.get('tools', []))}, thinking: {thinking_enabled}" + f"[PROVIDER] Anthropic API call - model: {params['model']}, messages: {len(params['messages'])}, system: {bool(params.get('system'))}, tools: {len(params.get('tools', []))}, thinking: {thinking_enabled}" ) # Emit llm:request event @@ -3825,14 +3967,17 @@ def _convert_to_chat_response(self, response: Any) -> ChatResponse: event_blocks.append(ThinkingContent(text=block.thinking)) # NOTE: Do NOT add thinking to text_accumulator - it's internal process, not response content elif block.type == "tool_use": + tool_name = self._oauth_tool_names.get( + block.name.lower(), block.name + ) content_blocks.append( - ToolCallBlock(id=block.id, name=block.name, input=block.input) + ToolCallBlock(id=block.id, name=tool_name, input=block.input) ) tool_calls.append( - ToolCall(id=block.id, name=block.name, arguments=block.input) + ToolCall(id=block.id, name=tool_name, arguments=block.input) ) event_blocks.append( - ToolCallContent(id=block.id, name=block.name, arguments=block.input) + ToolCallContent(id=block.id, name=tool_name, arguments=block.input) ) elif block.type == "web_search_tool_result": # Handle native web search results from Anthropic diff --git a/pyproject.toml b/pyproject.toml index 6420277..98ffb15 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -12,6 +12,10 @@ dependencies = [ "anthropic>=0.96.0", ] +[project.scripts] +amplifier-anthropic-login = "amplifier_anthropic_oauth.login:main" +amplifier-claude-login = "amplifier_anthropic_oauth.login:main" + [project.entry-points."amplifier.modules"] provider-anthropic = "amplifier_module_provider_anthropic:mount" @@ -27,6 +31,7 @@ package = true [tool.hatch.build.targets.wheel] packages = [ "amplifier_module_provider_anthropic", + "amplifier_anthropic_oauth", ] [tool.hatch.metadata] @@ -34,8 +39,12 @@ allow-direct-references = true [tool.pytest.ini_options] testpaths = ["tests"] -addopts = "--import-mode=importlib" +addopts = "--import-mode=importlib -m 'not live_oauth'" asyncio_mode = "strict" +markers = [ + "live_oauth: opt-in tests that call Anthropic with stored OAuth credentials", + "local_header_capture: opt-in local capture of installed Claude Code HTTPS headers", +] [dependency-groups] dev = [ diff --git a/tests/test_claude_header_capture.py b/tests/test_claude_header_capture.py new file mode 100644 index 0000000..90e1b8c --- /dev/null +++ b/tests/test_claude_header_capture.py @@ -0,0 +1,265 @@ +"""Optional local capture of the headers emitted by an installed ``claude -p``.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path +import queue +import shutil +import socketserver +import ssl +import subprocess +import threading + +import pytest + +from amplifier_anthropic_oauth.auth import OAUTH_BETAS, oauth_request_headers + + +class _CaptureProxy(socketserver.ThreadingTCPServer): + allow_reuse_address = True + daemon_threads = True + + def __init__(self, server_address, handler, tls_context: ssl.SSLContext): + super().__init__(server_address, handler) + self.tls_context = tls_context + self.captured: queue.Queue[dict[str, str]] = queue.Queue() + self.events: queue.Queue[str] = queue.Queue() + + +class _ConnectHandler(socketserver.BaseRequestHandler): + """Minimal CONNECT proxy which records only redacted request headers.""" + + def handle(self) -> None: + connection = self.request + connection.settimeout(5) + connect_request = self._read_headers(connection) + first_line = connect_request.split(b"\r\n", 1)[0] + self.server.events.put(first_line.decode(errors="replace")) + if first_line != b"CONNECT api.anthropic.com:443 HTTP/1.1": + connection.sendall(b"HTTP/1.1 502 Bad Gateway\r\n\r\n") + return + + connection.sendall(b"HTTP/1.1 200 Connection Established\r\n\r\n") + try: + self.server.events.put("tls-start") + tls = self.server.tls_context.wrap_socket(connection, server_side=True) + self.server.events.put("tls-ok") + request_head = self._read_headers(tls) + self.server.events.put( + request_head.split(b"\r\n", 1)[0].decode(errors="replace") + ) + lines = request_head.decode("iso-8859-1").split("\r\n") + request_line = lines[0].split(" ") + if ( + len(request_line) < 2 + or request_line[1].split("?", 1)[0] != "/v1/messages" + ): + self._respond(tls) + return + + raw_headers: dict[str, str] = {} + for line in lines[1:]: + if ":" not in line: + continue + name, value = line.split(":", 1) + raw_headers[name.strip().lower()] = value.strip() + + authorization = raw_headers.get("authorization", "") + # Never retain or write the credential. Only preserve its scheme. + normalized = { + "authorization": ( + "Bearer " + if authorization.startswith("Bearer ") + else "" + ), + "user-agent": raw_headers.get("user-agent", ""), + "x-app": raw_headers.get("x-app", ""), + "anthropic-beta": raw_headers.get("anthropic-beta", ""), + "anthropic-dangerous-direct-browser-access": raw_headers.get( + "anthropic-dangerous-direct-browser-access", "" + ), + } + self.server.captured.put(normalized) + self._respond(tls) + except (OSError, ssl.SSLError, TimeoutError) as exc: + self.server.events.put(f"{type(exc).__name__}: {exc}") + return + + @staticmethod + def _read_headers(connection) -> bytes: + data = bytearray() + while b"\r\n\r\n" not in data and len(data) < 128 * 1024: + chunk = connection.recv(4096) + if not chunk: + break + data.extend(chunk) + return bytes(data) + + @staticmethod + def _respond(connection) -> None: + body = json.dumps( + { + "type": "error", + "error": {"type": "authentication_error", "message": "captured"}, + } + ).encode() + connection.sendall( + b"HTTP/1.1 401 Unauthorized\r\n" + b"Content-Type: application/json\r\n" + + f"Content-Length: {len(body)}\r\nConnection: close\r\n\r\n".encode() + + body + ) + + +def _generate_ca_and_leaf(tmp_path: Path) -> tuple[Path, ssl.SSLContext]: + openssl = shutil.which("openssl") + if not openssl: + pytest.skip("openssl is required for local Claude header capture") + + ca_key = tmp_path / "ca.key" + ca_cert = tmp_path / "ca.pem" + leaf_key = tmp_path / "leaf.key" + leaf_csr = tmp_path / "leaf.csr" + leaf_cert = tmp_path / "leaf.pem" + extensions = tmp_path / "leaf.ext" + extensions.write_text( + "subjectAltName=DNS:api.anthropic.com\n" + "basicConstraints=critical,CA:FALSE\n" + "keyUsage=critical,digitalSignature,keyEncipherment\n" + "extendedKeyUsage=serverAuth\n" + ) + + commands = [ + [ + openssl, + "req", + "-x509", + "-newkey", + "rsa:2048", + "-nodes", + "-keyout", + str(ca_key), + "-out", + str(ca_cert), + "-days", + "1", + "-subj", + "/CN=Amplifier Claude Header Capture CA", + "-addext", + "basicConstraints=critical,CA:TRUE", + "-addext", + "keyUsage=critical,keyCertSign,cRLSign", + ], + [ + openssl, + "req", + "-newkey", + "rsa:2048", + "-nodes", + "-keyout", + str(leaf_key), + "-out", + str(leaf_csr), + "-subj", + "/CN=api.anthropic.com", + ], + [ + openssl, + "x509", + "-req", + "-in", + str(leaf_csr), + "-CA", + str(ca_cert), + "-CAkey", + str(ca_key), + "-CAcreateserial", + "-out", + str(leaf_cert), + "-days", + "1", + "-extfile", + str(extensions), + ], + ] + for command in commands: + subprocess.run(command, check=True, capture_output=True) + + context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + context.load_cert_chain(leaf_cert, leaf_key) + context.set_alpn_protocols(["http/1.1"]) + return ca_cert, context + + +@pytest.mark.local_header_capture +@pytest.mark.skipif(bool(os.environ.get("CI")), reason="local credential-bearing test") +def test_claude_p_stable_headers_match_provider(tmp_path): + """Capture a real local Claude request and compare its stable OAuth headers.""" + claude = shutil.which("claude") + if not claude: + pytest.skip("Claude Code is not installed") + if not (Path.home() / ".claude" / ".credentials.json").exists(): + pytest.skip("Claude Code credentials were not found") + + ca_cert, tls_context = _generate_ca_and_leaf(tmp_path) + proxy = _CaptureProxy(("127.0.0.1", 0), _ConnectHandler, tls_context) + thread = threading.Thread(target=proxy.serve_forever, daemon=True) + thread.start() + + proxy_url = f"http://127.0.0.1:{proxy.server_address[1]}" + env = os.environ.copy() + env.update( + { + "HTTPS_PROXY": proxy_url, + "https_proxy": proxy_url, + "NO_PROXY": "", + "no_proxy": "", + "NODE_EXTRA_CA_CERTS": str(ca_cert), + "NODE_USE_SYSTEM_CA": "1", + } + ) + for name in ("ANTHROPIC_API_KEY", "ANTHROPIC_AUTH_TOKEN", "ANTHROPIC_BASE_URL"): + env.pop(name, None) + + process = subprocess.Popen( + [claude, "-p", "Reply exactly OK", "--no-session-persistence"], + cwd=tmp_path, + env=env, + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.PIPE, + text=True, + ) + try: + try: + captured = proxy.captured.get(timeout=20) + except queue.Empty: + events = list(proxy.events.queue) + process.poll() + pytest.fail( + "Claude did not send a capturable request through the local proxy; " + f"returncode={process.returncode}, proxy_events={events}" + ) + finally: + process.terminate() + try: + process.wait(timeout=3) + except subprocess.TimeoutExpired: + process.kill() + process.wait(timeout=3) + proxy.shutdown() + proxy.server_close() + thread.join(timeout=3) + + expected = oauth_request_headers() + assert captured["authorization"] == "Bearer " + assert captured["user-agent"] == expected["User-Agent"] + assert captured["x-app"] == expected["x-app"] + assert captured["anthropic-dangerous-direct-browser-access"] == expected[ + "anthropic-dangerous-direct-browser-access" + ] + assert set(OAUTH_BETAS).issubset( + set(captured["anthropic-beta"].split(",")) + ) diff --git a/tests/test_claude_header_parity.py b/tests/test_claude_header_parity.py new file mode 100644 index 0000000..b8d5b13 --- /dev/null +++ b/tests/test_claude_header_parity.py @@ -0,0 +1,57 @@ +"""Drift checks against an installed Claude Code executable. + +Claude Code does not expose its final request headers, and OAuth traffic cannot +safely be redirected to a plaintext recorder. This test checks the observable +identity contract in the installed executable. The live integration tests then +verify that Anthropic accepts the resulting provider request. +""" + +from pathlib import Path +import re +import shutil +import subprocess + +import pytest + +from amplifier_anthropic_oauth.auth import ( + AUTHORIZE_URL, + CLIENT_ID, + OAUTH_BETAS, + TOKEN_URL, + installed_claude_code_version, + oauth_request_headers, +) + + +def test_installed_claude_code_oauth_header_contract(): + executable = shutil.which("claude") + if not executable: + pytest.skip("Claude Code is not installed") + + resolved = Path(executable).resolve() + binary = resolved.read_bytes() + headers = oauth_request_headers() + version = installed_claude_code_version() + + reported = subprocess.run( + [executable, "--version"], + capture_output=True, + text=True, + timeout=5, + check=True, + ).stdout + assert re.search(rf"\b{re.escape(version)}\b", reported) + expected_user_agent = f"claude-cli/{version} (external, sdk-cli)" + assert headers["User-Agent"] == expected_user_agent + assert b"claude-cli/" in binary + assert b"(external, " in binary + assert b"x-app" in binary + assert AUTHORIZE_URL.encode() in binary + assert TOKEN_URL.encode() in binary + assert CLIENT_ID.encode() in binary + + for beta in OAUTH_BETAS: + assert beta.encode() in binary, ( + f"Installed Claude Code no longer contains {beta}; inspect its " + "request contract before updating OAUTH_BETAS" + ) diff --git a/tests/test_oauth.py b/tests/test_oauth.py new file mode 100644 index 0000000..f4b627b --- /dev/null +++ b/tests/test_oauth.py @@ -0,0 +1,314 @@ +"""Tests for Anthropic subscription OAuth and native tool transport.""" + +import asyncio +import json +from io import BytesIO +import os +import subprocess +import sys +from urllib.error import HTTPError + +from amplifier_core.message_models import ToolSpec +from anthropic.types import Message as AnthropicMessage +import pytest + +from amplifier_module_provider_anthropic import AnthropicProvider +import amplifier_anthropic_oauth.auth as auth_module +import amplifier_anthropic_oauth.login as login_module +from amplifier_anthropic_oauth.auth import ( + AnthropicAuth, + AnthropicAuthError, + AnthropicAuthManager, + OAUTH_BETAS, + oauth_request_headers, + read_credentials, + refresh_oauth_credentials, + write_credentials, +) + + +def test_browser_callback_finishes_with_partial_terminal_input(tmp_path, monkeypatch): + """A readable partial stdin value must not block callback completion.""" + read_fd, write_fd = os.pipe() + stdin = os.fdopen(read_fd, "r") + os.write(write_fd, b"partial input without a newline") + + monkeypatch.setattr(login_module.sys, "stdin", stdin) + monkeypatch.setattr(login_module.webbrowser, "open", lambda url: False) + monkeypatch.setattr(login_module, "generate_pkce", lambda: ("verifier", "challenge")) + monkeypatch.setattr(login_module, "authorization_url", lambda *_: "https://example.test") + monkeypatch.setattr( + login_module, + "exchange_authorization_code", + lambda code, state, verifier: {"access": code}, + ) + monkeypatch.setattr(login_module, "default_auth_path", lambda: tmp_path / "auth.json") + saved = {} + monkeypatch.setattr( + login_module, + "write_credentials", + lambda path, credentials: saved.update(path=path, credentials=credentials), + ) + + original_start_server = asyncio.start_server + servers = [] + + async def start_test_server(callback, host, port): + server = await original_start_server(callback, host, 0) + servers.append(server) + return server + + monkeypatch.setattr(login_module.asyncio, "start_server", start_test_server) + + async def run_login_and_callback(): + task = asyncio.create_task(login_module.login()) + while not servers: + await asyncio.sleep(0) + port = servers[0].sockets[0].getsockname()[1] + reader, writer = await asyncio.open_connection("127.0.0.1", port) + writer.write( + b"GET /callback?code=test-code&state=verifier HTTP/1.1\r\n" + b"Host: localhost\r\n\r\n" + ) + await writer.drain() + await reader.read() + writer.close() + await writer.wait_closed() + await asyncio.wait_for(task, timeout=2) + + try: + asyncio.run(run_login_and_callback()) + assert os.get_blocking(read_fd) is True + finally: + stdin.close() + os.close(write_fd) + + assert saved == { + "path": tmp_path / "auth.json", + "credentials": {"access": "test-code"}, + } + + +def test_login_module_does_not_import_amplifier_core(): + code = """ +import builtins +original_import = builtins.__import__ +def guarded_import(name, *args, **kwargs): + if name == 'amplifier_core' or name.startswith('amplifier_core.'): + raise AssertionError('standalone login imported amplifier_core') + return original_import(name, *args, **kwargs) +builtins.__import__ = guarded_import +import amplifier_anthropic_oauth.login +""" + subprocess.run([sys.executable, "-c", code], check=True) + + +def test_token_exchange_uses_claude_identity_headers(monkeypatch): + captured = {} + + class Response: + def __enter__(self): + return self + + def __exit__(self, *args): + return False + + def read(self): + return b'{"ok": true}' + + def fake_urlopen(request, timeout): + captured["request"] = request + captured["timeout"] = timeout + return Response() + + monkeypatch.setenv("AMPLIFIER_CLAUDE_CODE_VERSION", "9.8.7") + monkeypatch.setattr(auth_module, "urlopen", fake_urlopen) + assert auth_module._post_json("https://example.test/token", {"code": "x"}) == { + "ok": True + } + headers = dict(captured["request"].header_items()) + assert headers["User-agent"] == "claude-cli/9.8.7 (external, sdk-cli)" + assert headers["Anthropic-beta"] == "oauth-2025-04-20" + assert headers["X-app"] == "cli" + + +def test_refresh_uses_oauth_endpoint_and_scopes(monkeypatch): + captured = {} + + def fake_post_json(url, body): + captured["url"] = url + captured["body"] = body + return { + "access_token": "new-access", + "refresh_token": "new-refresh", + "expires_in": 3600, + } + + monkeypatch.setattr(auth_module, "_post_json", fake_post_json) + refreshed = refresh_oauth_credentials({"refresh": "old-refresh"}) + assert captured["url"] == auth_module.TOKEN_URL + assert captured["body"] == { + "grant_type": "refresh_token", + "client_id": auth_module.CLIENT_ID, + "refresh_token": "old-refresh", + "scope": auth_module.SCOPES, + } + assert refreshed["access"] == "new-access" + assert refreshed["refresh"] == "new-refresh" + + +def test_token_exchange_error_includes_response_body(monkeypatch): + def fake_urlopen(request, timeout): + raise HTTPError( + request.full_url, + 403, + "Forbidden", + {}, + BytesIO(b'{"error":"invalid_request"}'), + ) + + monkeypatch.setattr(auth_module, "urlopen", fake_urlopen) + with pytest.raises(AnthropicAuthError, match="invalid_request"): + auth_module._post_json("https://example.test/token", {"code": "x"}) + + +def test_oauth_headers_have_claude_code_identity(monkeypatch): + monkeypatch.setenv("AMPLIFIER_CLAUDE_CODE_VERSION", "9.8.7") + headers = oauth_request_headers() + assert headers["x-app"] == "cli" + assert headers["User-Agent"] == "claude-cli/9.8.7 (external, sdk-cli)" + assert set(headers["anthropic-beta"].split(",")) == set(OAUTH_BETAS) + assert headers["anthropic-dangerous-direct-browser-access"] == "true" + + +def test_credentials_are_written_atomically_with_private_permissions(tmp_path): + path = tmp_path / "auth.json" + credentials = { + "type": "oauth", + "access": "sk-ant-oat-access", + "refresh": "refresh", + "expires": 9999999999999, + } + write_credentials(path, credentials) + assert read_credentials(path) == credentials + assert os.stat(path).st_mode & 0o777 == 0o600 + + +def test_auth_manager_prefers_oauth_over_api_key(tmp_path, monkeypatch): + monkeypatch.delenv("ANTHROPIC_OAUTH_TOKEN", raising=False) + monkeypatch.setenv("ANTHROPIC_API_KEY", "api-key") + path = tmp_path / "auth.json" + write_credentials( + path, + { + "type": "oauth", + "access": "sk-ant-oat-access", + "refresh": "refresh", + "expires": 9999999999999, + }, + ) + auth = asyncio.run(AnthropicAuthManager(path=path).get_auth()) + assert auth == AnthropicAuth("sk-ant-oat-access", oauth=True) + + +def test_api_key_config_is_optional_for_oauth_users(): + provider = AnthropicProvider( + "sk-ant-oat-test", + initial_auth=AnthropicAuth("sk-ant-oat-test", oauth=True), + ) + api_key_field = next( + field for field in provider.get_info().config_fields if field.id == "api_key" + ) + assert api_key_field.required is False + + +@pytest.mark.parametrize("api_key", [None, ""]) +def test_direct_provider_instance_discovers_stored_oauth(tmp_path, api_key): + path = tmp_path / "auth.json" + write_credentials( + path, + { + "type": "oauth", + "access": "sk-ant-oat-discovered", + "refresh": "refresh", + "expires": 9999999999999, + }, + ) + provider = AnthropicProvider(api_key=api_key, config={"auth_file": str(path)}) + asyncio.run(provider._refresh_auth()) + assert provider._auth_state == AnthropicAuth( + "sk-ant-oat-discovered", oauth=True + ) + assert provider._default_headers["x-app"] == "cli" + assert "oauth-2025-04-20" in provider._default_headers["anthropic-beta"] + assert provider.client.api_key is None + assert provider.client.auth_token == "sk-ant-oat-discovered" + + +def test_oauth_client_uses_bearer_auth_and_identity_headers(): + provider = AnthropicProvider( + "sk-ant-oat-test", + initial_auth=AnthropicAuth("sk-ant-oat-test", oauth=True), + ) + client = provider.client + assert client.api_key is None + assert client.auth_token == "sk-ant-oat-test" + assert client.default_headers["x-app"] == "cli" + assert "oauth-2025-04-20" in client.default_headers["anthropic-beta"] + + +def test_native_tools_are_sent_structurally(): + provider = AnthropicProvider( + "sk-ant-oat-test", + initial_auth=AnthropicAuth("sk-ant-oat-test", oauth=True), + ) + tools = provider._convert_tools_from_request( + [ + ToolSpec( + name="read", + description="Read a file", + parameters={ + "type": "object", + "properties": {"path": {"type": "string"}}, + "required": ["path"], + }, + ) + ] + ) + params = { + "messages": [{"role": "user", "content": "Read the README"}], + "tools": tools, + } + provider._apply_oauth_request_contract(params) + + response = AnthropicMessage.model_validate( + { + "id": "msg_test", + "type": "message", + "role": "assistant", + "model": "claude-sonnet-4-6", + "content": [ + { + "type": "tool_use", + "id": "toolu_1", + "name": "Read", + "input": {"path": "README.md"}, + } + ], + "stop_reason": "tool_use", + "stop_sequence": None, + "usage": {"input_tokens": 10, "output_tokens": 5}, + } + ) + result = provider._convert_to_chat_response(response) + + assert params["tools"][0]["name"] == "Read" + assert params["tools"][0]["input_schema"]["required"] == ["path"] + assert params["messages"] == [ + {"role": "user", "content": "Read the README"} + ] + assert "" not in json.dumps(params["messages"]) + assert "[tool]:" not in json.dumps(params["messages"]) + assert params["system"][0]["text"].startswith("You are Claude Code") + assert result.tool_calls[0].name == "read" + assert result.tool_calls[0].arguments == {"path": "README.md"} diff --git a/tests/test_oauth_transport.py b/tests/test_oauth_transport.py new file mode 100644 index 0000000..1117106 --- /dev/null +++ b/tests/test_oauth_transport.py @@ -0,0 +1,149 @@ +"""Transport-level and opt-in live tests for Claude subscription OAuth.""" + +from __future__ import annotations + +from amplifier_core.message_models import ChatRequest, Message, ToolSpec +from anthropic import AsyncAnthropic +import httpx +import pytest + +from amplifier_anthropic_oauth.auth import ( + AnthropicAuth, + OAUTH_BETAS, + default_auth_path, + oauth_request_headers, +) +from amplifier_module_provider_anthropic import AnthropicProvider + + +def _assert_oauth_headers(headers: httpx.Headers) -> None: + expected = oauth_request_headers() + assert headers["authorization"] == "Bearer sk-ant-oat-transport-test" + assert "x-api-key" not in headers + assert headers["User-Agent"] == expected["User-Agent"] + assert headers["x-app"] == expected["x-app"] + assert set(OAUTH_BETAS).issubset(set(headers["anthropic-beta"].split(","))) + + +@pytest.mark.asyncio +async def test_models_and_messages_emit_oauth_headers(monkeypatch): + """Inspect the final HTTP requests emitted by the Anthropic SDK.""" + monkeypatch.setenv("AMPLIFIER_CLAUDE_CODE_VERSION", "9.8.7") + requests: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + requests.append(request) + path = request.url.path + if path == "/v1/models": + return httpx.Response( + 200, + json={ + "data": [ + { + "type": "model", + "id": "claude-haiku-4-5", + "display_name": "Claude Haiku 4.5", + "created_at": "2025-10-01T00:00:00Z", + } + ], + "has_more": False, + "first_id": "claude-haiku-4-5", + "last_id": "claude-haiku-4-5", + }, + ) + if path.startswith("/v1/models/"): + return httpx.Response( + 200, + json={ + "type": "model", + "id": "claude-haiku-4-5", + "display_name": "Claude Haiku 4.5", + "created_at": "2025-10-01T00:00:00Z", + }, + ) + if path == "/v1/messages": + return httpx.Response( + 200, + json={ + "id": "msg_oauth_transport", + "type": "message", + "role": "assistant", + "model": "claude-haiku-4-5", + "content": [{"type": "text", "text": "ok"}], + "stop_reason": "end_turn", + "stop_sequence": None, + "usage": {"input_tokens": 1, "output_tokens": 1}, + }, + ) + return httpx.Response(404, json={"error": {"message": path}}) + + auth = AnthropicAuth("sk-ant-oat-transport-test", oauth=True) + provider = AnthropicProvider( + auth.token, + config={ + "default_model": "claude-haiku-4-5", + "use_streaming": False, + "max_retries": 0, + }, + initial_auth=auth, + ) + sdk_client = provider.client + await sdk_client._client.aclose() + sdk_client._client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + + models = await provider.list_models() + assert models + response = await provider.complete( + ChatRequest( + messages=[Message(role="user", content="Reply with ok")], + metadata={"stream": False}, + max_output_tokens=64, + ) + ) + assert response.text == "ok" + + assert any(request.url.path == "/v1/models" for request in requests) + assert any(request.url.path == "/v1/messages" for request in requests) + for request in requests: + _assert_oauth_headers(request.headers) + + +@pytest.mark.live_oauth +@pytest.mark.asyncio +async def test_live_oauth_models_and_native_tool_call(): + """Opt-in smoke test against Anthropic using the stored OAuth credential.""" + auth_file = default_auth_path() + provider = AnthropicProvider( + config={ + "auth_file": str(auth_file), + "default_model": "claude-haiku-4-5", + "use_streaming": False, + "max_retries": 0, + "max_tokens": 64, + } + ) + + models = await provider.list_models() + assert any(model.id.startswith("claude-haiku-4-5") for model in models) + + result = await provider.complete( + ChatRequest( + messages=[Message(role="user", content="Call oauth_probe with value ok")], + tools=[ + ToolSpec( + name="oauth_probe", + description="OAuth integration probe; always call this tool", + parameters={ + "type": "object", + "properties": {"value": {"type": "string"}}, + "required": ["value"], + }, + ) + ], + metadata={"stream": False}, + max_output_tokens=64, + ), + tool_choice={"type": "tool", "name": "oauth_probe"}, + ) + assert result.tool_calls + assert result.tool_calls[0].name == "oauth_probe"