File size: 3,353 Bytes
dbb04e4 c3a3710 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 | """
MnemoCore API Adapter
=====================
HTTP client adapter for communicating with MnemoCore API server.
"""
from typing import Any, Dict, Optional
import requests
from mnemocore.core.exceptions import MnemoCoreError
class MnemoCoreAPIError(MnemoCoreError):
"""
Exception raised when API communication fails.
Attributes:
status_code: HTTP status code if available (None for network errors).
"""
def __init__(self, message: str, status_code: Optional[int] = None, context: Optional[dict] = None):
ctx = context or {}
if status_code is not None:
ctx["status_code"] = status_code
super().__init__(message, ctx)
self.status_code = status_code
class MnemoCoreAPIAdapter:
def __init__(self, base_url: str, api_key: str, timeout_seconds: int = 15):
self.base_url = base_url.rstrip("/")
self.api_key = api_key
self.timeout_seconds = timeout_seconds
def _request(self, method: str, path: str, payload: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
url = f"{self.base_url}{path}"
headers = {
"X-API-Key": self.api_key,
"Content-Type": "application/json",
}
try:
response = requests.request(
method=method,
url=url,
json=payload,
headers=headers,
timeout=self.timeout_seconds,
)
except requests.RequestException as exc:
raise MnemoCoreAPIError(f"Upstream request failed: {exc}") from exc
if response.status_code >= 400:
try:
details = response.json()
except ValueError:
details = {"detail": response.text}
raise MnemoCoreAPIError(
f"Upstream error ({response.status_code}): {details}",
status_code=response.status_code,
)
try:
return response.json()
except ValueError as exc:
raise MnemoCoreAPIError("Upstream returned non-JSON response") from exc
def store(self, payload: Dict[str, Any]) -> Dict[str, Any]:
return self._request("POST", "/store", payload)
def query(self, payload: Dict[str, Any]) -> Dict[str, Any]:
return self._request("POST", "/query", payload)
def get_memory(self, memory_id: str) -> Dict[str, Any]:
return self._request("GET", f"/memory/{memory_id}")
def delete_memory(self, memory_id: str) -> Dict[str, Any]:
return self._request("DELETE", f"/memory/{memory_id}")
def stats(self) -> Dict[str, Any]:
return self._request("GET", "/stats")
def health(self) -> Dict[str, Any]:
return self._request("GET", "/health")
# --- Phase 5: Cognitive Client Adapters ---
def observe_context(self, payload: Dict[str, Any]) -> Dict[str, Any]:
return self._request("POST", "/wm/observe", payload)
def get_working_context(self, agent_id: str, limit: int = 16) -> Dict[str, Any]:
return self._request("GET", f"/wm/context/{agent_id}?limit={limit}")
def start_episode(self, payload: Dict[str, Any]) -> Dict[str, Any]:
return self._request("POST", "/episodes/start", payload)
|