""" Coppermind API client — zero-dependency HTTP client using only stdlib. Talks to the Coppermind API (backed by SurrealDB) for memory retrieval, search, and ingestion. Works with both the hosted gateway (api.coppermindapi.com) and local gateway (localhost:8787). """ import json import logging import os import urllib.request import urllib.error import urllib.parse from typing import Any, Optional logger = logging.getLogger("coppermind") # --------------------------------------------------------------------------- # Configuration # --------------------------------------------------------------------------- API_KEY = os.environ.get("COPPERMIND_API_KEY", "") BASE_URL = os.environ.get("COPPERMIND_BASE_URL", "https://api.coppermindapi.com") USER_ID = os.environ.get("COPPERMIND_USER_ID", "default") TIMEOUT = int(os.environ.get("COPPERMIND_TIMEOUT", "10")) # Cached parsed URL to avoid repeated parsing on every is_local_runtime() call _cached_parsed_url = None def _parsed_base_url(): global _cached_parsed_url if _cached_parsed_url is None: try: _cached_parsed_url = urllib.parse.urlparse(BASE_URL) except Exception: _cached_parsed_url = None return _cached_parsed_url def is_local_runtime() -> bool: parsed = _parsed_base_url() if parsed is None: return False if parsed.scheme in {"local"}: return True hostname = (parsed.hostname or "").strip().lower() return hostname in {"127.0.0.1", "localhost", "::1"} def requires_api_key() -> bool: return not is_local_runtime() class CoppermindError(Exception): """Error from the Coppermind API.""" def __init__(self, message: str, status: int | None = None, code: str | None = None): super().__init__(message) self.status = status self.code = code # --------------------------------------------------------------------------- # Low-level HTTP helper # --------------------------------------------------------------------------- def _request( method: str, path: str, body: dict | None = None, params: dict | None = None, ) -> dict[str, Any]: """Make an HTTP request to the Coppermind API. Uses only stdlib urllib — no third-party dependencies required. """ url = f"{BASE_URL.rstrip('/')}{path}" if params: query = urllib.parse.urlencode( {k: v for k, v in params.items() if v is not None} ) if query: url = f"{url}?{query}" headers = { "Content-Type": "application/json", } if API_KEY.strip(): headers["Authorization"] = f"Bearer {API_KEY}" data = json.dumps(body).encode("utf-8") if body else None req = urllib.request.Request(url, data=data, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=TIMEOUT) as resp: resp_body = resp.read().decode("utf-8") if resp_body: return json.loads(resp_body) return {"success": True} except urllib.error.HTTPError as e: resp_body = e.read().decode("utf-8", errors="replace") try: error_data = json.loads(resp_body) msg = ( error_data.get("error", {}).get("message") or error_data.get("message") or resp_body ) except (json.JSONDecodeError, AttributeError): msg = resp_body or str(e) raise CoppermindError(msg, status=e.code, code="API_ERROR") from e except urllib.error.URLError as e: raise CoppermindError( f"Cannot reach Coppermind API at {BASE_URL}: {e.reason}", code="CONNECTION_ERROR", ) from e except Exception as e: raise CoppermindError(f"Unexpected error: {e}", code="UNKNOWN") from e # --------------------------------------------------------------------------- # Public API # --------------------------------------------------------------------------- def search_memory( query: str, user_id: str | None = None, limit: int = 5, min_score: float | None = None, tags: list[str] | None = None, ) -> dict[str, Any]: """Semantic search across memories. Args: query: Natural language search query. user_id: User ID scope (defaults to COPPERMIND_USER_ID env var). limit: Maximum number of results. min_score: Minimum relevance score (0-1). tags: Optional tags to filter by. Returns: Dict with "context" (string) or "fragments" (list). """ params: dict[str, Any] = { "user_id": user_id or USER_ID, "query": query, "limit": str(limit), } if min_score is not None: params["min_score"] = str(min_score) if tags: params["tags"] = ",".join(tags) return _request("GET", "/v1/memory/search", params=params) def get_memory(user_id: str | None = None, query: str | None = None) -> dict[str, Any]: """Retrieve memory context for a user. Args: user_id: User ID scope. query: Optional query for relevance-scored retrieval. Returns: Dict with "context" (formatted memory string). """ params: dict[str, Any] = {"user_id": user_id or USER_ID} if query: params["query"] = query return _request("GET", "/v1/memory", params=params) def ingest_memory( content: str, user_id: str | None = None, tags: list[str] | None = None, metadata: dict | None = None, session_id: str | None = None, project_id: str | None = None, task_id: str | None = None, ) -> dict[str, Any]: """Store new memory content. Args: content: The text content to store as memory. user_id: User ID scope. tags: Optional tags for categorization. metadata: Optional metadata dict. session_id: Optional stable session scope for related turns. project_id: Optional project scope identifier. task_id: Optional task scope identifier. Returns: Dict with "success" (bool) and "inserted" (int). """ body: dict[str, Any] = { "userId": user_id or USER_ID, **({"sessionId": session_id} if session_id else {}), **({"projectId": project_id} if project_id else {}), **({"taskId": task_id} if task_id else {}), "data": [ { "content": content, **({"tags": tags} if tags else {}), **({"metadata": metadata} if metadata else {}), } ], } return _request("POST", "/v1/ingest", body=body) def get_stats(user_id: str | None = None) -> dict[str, Any]: """Get memory statistics. Args: user_id: User ID scope. Returns: Dict with totalEpisodes, totalTokens, etc. """ return _request("GET", "/v1/stats", params={"user_id": user_id or USER_ID}) def health_check() -> dict[str, Any]: """Check API health.""" return _request("GET", "/health")