"""Claude provider — accesses claude.ai internal web API.""" import logging import os from src.providers.base import BaseProvider, ProviderError logger = logging.getLogger(__name__) BASE_URL = "https://claude.ai/api" class ClaudeProvider(BaseProvider): """Provider for Claude conversations via the internal web API. Authentication: Cookie: sessionKey= Token: sessionKey cookie value from claude.ai. Typical validity: ~30 days (opaque; expiry cannot be decoded client-side). """ provider_name = "claude" def __init__(self, session_key: str | None = None) -> None: super().__init__() key = session_key or os.getenv("CLAUDE_SESSION_KEY", "").strip() if not key: raise ProviderError( self.provider_name, "init", RuntimeError( "CLAUDE_SESSION_KEY is not set. " "Run 'python -m src.main auth' to configure it." ), ) # Set sessionKey in the cookie jar (not as a raw header string) self._session.cookies.set("sessionKey", key, domain="claude.ai", path="/") self._session.headers.update( { "Referer": "https://claude.ai/", "Origin": "https://claude.ai", } ) self._org_id: str | None = None # cached per session logger.debug("[claude] Session initialised (key: [REDACTED])") def _handle_401(self) -> None: msg = ( "[claude] Authentication failed (401 Unauthorized). " "Your sessionKey has likely expired (~30 day lifetime). " "Note: Claude session keys are opaque — a 401 is the only expiry signal. " "To refresh: open claude.ai in Chrome → F12 → Application → Cookies " "→ find 'sessionKey' → copy the value. " "Then run 'python -m src.main auth' or update CLAUDE_SESSION_KEY in .env." ) logger.error(msg) raise ProviderError( self.provider_name, "authentication", RuntimeError("401 Unauthorized — Claude session key expired"), ) def _get_org_id(self) -> str: """Fetch and cache the organization ID (required for all Claude API calls).""" if self._org_id: return self._org_id logger.debug("[claude] Bootstrapping: fetching organization ID") url = f"{BASE_URL}/organizations" try: data = self._make_request("GET", url) except ProviderError: raise except Exception as e: raise ProviderError(self.provider_name, "get_org_id", e) from e if not isinstance(data, list) or not data: self._warn_unexpected_schema("get_org_id", "organizations array") raise ProviderError( self.provider_name, "get_org_id", RuntimeError("organizations endpoint returned empty or unexpected response"), ) org = data[0] org_id = org.get("uuid") or org.get("id") if not org_id: self._warn_unexpected_schema("get_org_id", "uuid/id") raise ProviderError( self.provider_name, "get_org_id", RuntimeError("Could not find organization ID in response"), ) self._org_id = org_id logger.debug("[claude] Got org_id: %s", org_id) return org_id def list_conversations(self, offset: int = 0, limit: int = 100) -> list[dict]: """Fetch one page of conversations. Claude's API may use cursor-based pagination. We attempt offset-based first (via query param); if the response includes a ``next_cursor`` field, a WARNING is logged indicating manual investigation is needed. """ org_id = self._get_org_id() url = f"{BASE_URL}/organizations/{org_id}/chat_conversations" params: dict = {"limit": limit} if offset > 0: params["offset"] = offset try: data = self._make_request("GET", url, params=params) except ProviderError: raise except Exception as e: raise ProviderError(self.provider_name, "list_conversations", e) from e # Handle both list and dict responses if isinstance(data, list): conversations = data elif isinstance(data, dict): conversations = data.get("conversations") or data.get("chats") or [] if not conversations and data: # Check for unexpected pagination mechanism if "next_cursor" in data or "cursor" in data or "next" in data: logger.warning( "[claude] API returned cursor-based pagination — " "only first page will be fetched. " "Please report this at GitHub Issues." ) else: self._warn_unexpected_schema("list_conversations", "root") return [] return conversations def get_conversation(self, conv_id: str) -> dict: """Fetch full conversation detail for a single ID.""" org_id = self._get_org_id() url = f"{BASE_URL}/organizations/{org_id}/chat_conversations/{conv_id}" try: data = self._make_request("GET", url) except ProviderError: raise except Exception as e: raise ProviderError(self.provider_name, "get_conversation", e) from e if not isinstance(data, dict): self._warn_unexpected_schema("get_conversation", "root") return {} return data def normalize_conversation(self, raw: dict) -> dict: """Transform Claude raw schema to the common normalized schema.""" conv_id = raw.get("uuid") or raw.get("id", "") title = raw.get("name") or raw.get("title") or "Untitled" created_at = raw.get("created_at") or raw.get("create_time") or "" updated_at = raw.get("updated_at") or raw.get("update_time") or "" # Project name — Claude may nest project info as project.name project_data = raw.get("project") or {} project: str | None = ( project_data.get("name") if isinstance(project_data, dict) else project_data ) or None # Messages raw_messages = raw.get("chat_messages") or raw.get("messages") or [] messages = [] for msg in raw_messages: role = _map_role(msg.get("sender") or msg.get("role", "")) if not role: continue # Content can be a string or a list of content blocks content_raw = msg.get("content") or msg.get("text") or "" content, skipped_types = _extract_claude_text(content_raw, conv_id) for ctype in skipped_types: logger.warning( "[claude] Skipping %s content in conversation %s " "— rich content not yet supported (see FUTURE.md)", ctype, conv_id[:8], ) timestamp = msg.get("created_at") or msg.get("timestamp") or None if content is None: logger.debug("[claude] Skipping empty message in conversation %s", conv_id[:8]) continue messages.append( { "role": role, "content": content, "content_type": "text", "timestamp": timestamp, } ) return { "id": conv_id, "title": title, "provider": "claude", "project": project, "created_at": created_at, "updated_at": updated_at, "message_count": len(messages), "messages": messages, } # --------------------------------------------------------------------------- # Internal helpers # --------------------------------------------------------------------------- def _map_role(sender: str) -> str | None: """Map Claude sender strings to normalized roles.""" mapping = { "human": "user", "user": "user", "assistant": "assistant", "claude": "assistant", "ai": "assistant", "system": "system", } return mapping.get(sender.lower()) if sender else None def _extract_claude_text( content: str | list | dict, conv_id: str ) -> tuple[str | None, list[str]]: """Extract plain text from a Claude content field. Returns: (text_or_None, list_of_skipped_content_types) """ skipped: list[str] = [] if isinstance(content, str): text = content.strip() return (text if text else None), skipped if isinstance(content, list): parts: list[str] = [] for block in content: if isinstance(block, str): parts.append(block) elif isinstance(block, dict): btype = block.get("type", "text") if btype == "text": t = block.get("text", "").strip() if t: parts.append(t) else: skipped.append(btype) text = "\n".join(parts).strip() return (text if text else None), skipped if isinstance(content, dict): btype = content.get("type", "text") if btype == "text": text = content.get("text", "").strip() return (text if text else None), skipped else: skipped.append(btype) return None, skipped return None, skipped