"""Mem.ai API client wrapper with async support."""
import logging
import os
from typing import Optional
from uuid import UUID
import httpx
from dotenv import load_dotenv
from .models import (
Collection,
CollectionCreate,
MemItRequest,
Note,
NoteCreate,
RequestResponse,
)
# Load environment variables
load_dotenv()
logger = logging.getLogger(__name__)
class MemAPIError(Exception):
"""Base exception for Mem.ai API errors."""
pass
class MemAuthenticationError(MemAPIError):
"""Raised when authentication fails."""
pass
class MemNotFoundError(MemAPIError):
"""Raised when a resource is not found."""
pass
class MemValidationError(MemAPIError):
"""Raised when request validation fails."""
pass
class MemClient:
"""Async client for Mem.ai API.
Provides a clean, type-safe interface to all Mem.ai API endpoints
with comprehensive error handling and automatic retries.
Args:
api_key: Mem.ai API key (defaults to MEM_API_KEY env variable)
base_url: API base URL (defaults to MEM_API_BASE_URL env or https://api.mem.ai/v2)
timeout: Request timeout in seconds (defaults to MEM_REQUEST_TIMEOUT env or 30)
"""
def __init__(
self,
api_key: Optional[str] = None,
base_url: Optional[str] = None,
timeout: float = 30.0,
):
self.api_key = api_key or os.getenv("MEM_API_KEY")
if not self.api_key:
raise MemAuthenticationError(
"MEM_API_KEY environment variable or api_key parameter is required"
)
self.base_url = (
base_url
or os.getenv("MEM_API_BASE_URL", "https://api.mem.ai/v2").rstrip("/")
)
timeout_env = os.getenv("MEM_REQUEST_TIMEOUT")
self.timeout = float(timeout_env) if timeout_env else timeout
self.client = httpx.AsyncClient(
base_url=self.base_url,
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
},
timeout=self.timeout,
)
logger.info(f"Initialized MemClient with base_url={self.base_url}")
async def __aenter__(self):
"""Async context manager entry."""
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
"""Async context manager exit."""
await self.close()
async def close(self):
"""Close the HTTP client."""
await self.client.aclose()
def _handle_error(self, response: httpx.Response) -> None:
"""Handle HTTP error responses.
Args:
response: HTTP response object
Raises:
MemAuthenticationError: For 401/403 errors
MemNotFoundError: For 404 errors
MemValidationError: For 400/422 errors
MemAPIError: For other errors
"""
if response.status_code == 401 or response.status_code == 403:
raise MemAuthenticationError(
f"Authentication failed: {response.text}"
)
elif response.status_code == 404:
raise MemNotFoundError(f"Resource not found: {response.text}")
elif response.status_code == 400 or response.status_code == 422:
raise MemValidationError(f"Validation error: {response.text}")
else:
raise MemAPIError(
f"API error (status {response.status_code}): {response.text}"
)
async def mem_it(
self,
input: str,
instructions: Optional[str] = None,
context: Optional[str] = None,
timestamp: Optional[str] = None,
) -> RequestResponse:
"""Save and intelligently process content with Mem It.
Args:
input: Content to save (text, HTML, markdown, etc.)
instructions: Optional processing instructions
context: Optional additional context
timestamp: Optional ISO 8601 timestamp
Returns:
RequestResponse with request_id
Raises:
MemAPIError: If the API request fails
"""
request = MemItRequest(
input=input,
instructions=instructions,
context=context,
timestamp=timestamp,
)
logger.debug(f"Mem It request: {len(input)} chars")
response = await self.client.post(
"/mem-it",
json=request.model_dump(exclude_none=True, mode="json"),
)
if response.status_code != 200:
self._handle_error(response)
return RequestResponse(**response.json())
async def create_note(
self,
content: str,
collection_ids: Optional[list[str]] = None,
collection_titles: Optional[list[str]] = None,
) -> Note:
"""Create a new note.
Args:
content: Markdown-formatted note content
collection_ids: Optional list of collection UUIDs
collection_titles: Optional list of collection titles
Returns:
Created Note object
Raises:
MemAPIError: If the API request fails
"""
# Convert string UUIDs to UUID objects if provided
uuid_collection_ids = None
if collection_ids:
uuid_collection_ids = [UUID(cid) for cid in collection_ids]
request = NoteCreate(
content=content,
collection_ids=uuid_collection_ids,
collection_titles=collection_titles,
)
logger.debug(f"Creating note: {len(content)} chars")
response = await self.client.post(
"/notes",
json=request.model_dump(exclude_none=True, mode="json"),
)
if response.status_code != 200:
self._handle_error(response)
return Note(**response.json())
async def read_note(self, note_id: str) -> Note:
"""Read a note by ID.
Args:
note_id: UUID of the note to read
Returns:
Note object
Raises:
MemNotFoundError: If the note doesn't exist
MemAPIError: If the API request fails
"""
logger.debug(f"Reading note: {note_id}")
response = await self.client.get(f"/notes/{note_id}")
if response.status_code != 200:
self._handle_error(response)
return Note(**response.json())
async def delete_note(self, note_id: str) -> RequestResponse:
"""Delete a note by ID.
Args:
note_id: UUID of the note to delete
Returns:
RequestResponse with request_id
Raises:
MemNotFoundError: If the note doesn't exist
MemAPIError: If the API request fails
"""
logger.debug(f"Deleting note: {note_id}")
response = await self.client.delete(f"/notes/{note_id}")
if response.status_code != 200:
self._handle_error(response)
return RequestResponse(**response.json())
async def create_collection(
self,
title: str,
description: Optional[str] = None,
) -> Collection:
"""Create a new collection.
Args:
title: Collection title
description: Optional markdown-formatted description
Returns:
Created Collection object
Raises:
MemAPIError: If the API request fails
"""
request = CollectionCreate(
title=title,
description=description,
)
logger.debug(f"Creating collection: {title}")
response = await self.client.post(
"/collections",
json=request.model_dump(exclude_none=True, mode="json"),
)
if response.status_code != 200:
self._handle_error(response)
return Collection(**response.json())
async def delete_collection(self, collection_id: str) -> RequestResponse:
"""Delete a collection by ID.
Args:
collection_id: UUID of the collection to delete
Returns:
RequestResponse with request_id
Raises:
MemNotFoundError: If the collection doesn't exist
MemAPIError: If the API request fails
"""
logger.debug(f"Deleting collection: {collection_id}")
response = await self.client.delete(f"/collections/{collection_id}")
if response.status_code != 200:
self._handle_error(response)
return RequestResponse(**response.json())
