pip install sovant

import os
from sovant import Sovant, MemoryCreate

# Initialize with basic config
client = Sovant(api_key=os.environ["SOVANT_API_KEY"])

# Create a memory
memory = client.memory_create(MemoryCreate(
    data="User prefers dark mode for all applications",
    type="preference",
    tags=["ui", "settings"]
))

print(f"Created: {memory['id']}")

from sovant import Sovant

client = Sovant(
    # Required
    api_key="sk_live_your_api_key_here",

    # Optional settings
    base_url="https://sovant.ai",  # Default, can override for testing
    timeout=30.0,                   # Request timeout in seconds (default: 30.0)
    max_retries=3,                  # Auto-retry attempts (default: 3)
)

from sovant import MemoryCreate

# Simple create
memory = client.memory_create(MemoryCreate(
    data="User completed onboarding tutorial",
    type="observation",
    tags=["onboarding", "completed"],
    metadata={"importance": 0.7},
    thread_id="thread-uuid-here"  # Optional
))

# Create with dict data
complex_memory = client.memory_create(MemoryCreate(
    data={
        "event": "onboarding_completed",
        "decisions": ["completed-tutorial"],
        "action_items": ["Send follow-up email"]
    },
    type="journal",
    tags=["milestone"]
))

memory = client.memory_get("memory-id-here")
print(memory["content"])
print(memory["type"])
print(memory["tags"])

updated = client.memory_update("memory-id-here", {
    "data": "Updated content",
    "tags": ["updated", "important"],
    "metadata": {"priority": "high"}
})

client.memory_delete("memory-id-here")
# Returns None on success

from sovant import SearchQuery

# Semantic search with natural language
results = client.memory_search(SearchQuery(
    query="what are my project preferences?",
    limit=10,
    type="preference",
    tags=["project"],
    from_date="2025-01-01T00:00:00Z",
    to_date="2025-12-31T23:59:59Z"
))

print(f"Found {results['total']} memories")
for r in results.get("results", []):
    print(f"- {r['content']} (relevance: {r.get('relevance_score', 0)})")

recall = client.memory_recall(
    query="what do you know about me?",
    limit=10,
    thread_id="thr_123"  # Optional: scope to a thread
)

# Response shape: { "results": [...], "total": N, "query_type": "hybrid" }
for mem in recall.get("results", []):
    print(mem["content"], mem.get("relevance"))

result = client.memory_create_batch([
    {
        "data": "Meeting with design team on Friday at 2 PM",
        "type": "task",
        "tags": ["meeting", "design"]
    },
    {
        "data": "User is located in San Francisco, California",
        "type": "observation",
        "tags": ["location", "personal"]
    },
    {
        "data": "Preferred programming language is Python",
        "type": "preference",
        "tags": ["development"]
    }
])

print(f"Created {result['summary']['successful']}/{result['summary']['total']} memories")

# Check individual results
for i, r in enumerate(result["results"]):
    if r["success"]:
        print(f"Memory {i}: {r['id']}")
    else:
        print(f"Memory {i} failed: {r['error']['message']}")

thread = client.threads_create(
    title="Customer Support Chat #1234",
    description="Issue with dashboard loading",
    metadata={"priority": "high"}
)

threads = client.threads_list(limit=20, offset=0)

thread = client.threads_get(
    "thread_id_here",
    include_memories=True,
    limit=50
)

updated = client.threads_update(
    "thread_id_here",
    title="Resolved: Dashboard Loading Issue",
    status="completed"
)

# Delete thread only (keeps memories)
client.threads_delete("thread_id_here")

# Delete thread AND all its memories
client.threads_delete("thread_id_here", delete_memories=True)

# Create a thread
thread = client.threads_create(
    title="Q1 Planning Session"
)

# Store memories in the thread
client.memory_create(MemoryCreate(
    data="Decided to launch in March",
    type="journal",
    thread_id=thread["id"]
))

client.memory_create(MemoryCreate(
    data="Budget approved: $50k",
    type="insight",
    thread_id=thread["id"]
))

# Recall from this thread specifically
recall = client.memory_recall(
    query="launch date",
    thread_id=thread["id"],
    limit=10
)
print(recall.get("results", []))

# Automatic retry on:
# - 429 (Rate Limit)
# - 5xx (Server Errors)
# - Network errors
# - Timeouts

client = Sovant(
    api_key="sk_live_...",
    max_retries=3,      # Will retry up to 3 times
)

# This will auto-retry if it hits rate limits or server errors
memory = client.memory_create(MemoryCreate(data="Test"))

from sovant import Sovant, SovantError, MemoryCreate

try:
    client.memory_create(MemoryCreate(data="Test"))
except SovantError as e:
    print(f"Error: {e}")
    print(f"Code: {e.code}")      # e.g. "NOT_FOUND", "HTTP_401"
    print(f"Status: {e.status}")  # e.g. 404, 401

    if e.status == 404:
        print("Memory not found")
    elif e.status == 401:
        print("Invalid API key")
    elif e.status == 429:
        print("Rate limited -- SDK retries automatically")

from typing import Dict, List, Any, Optional
from sovant import SovantError

# Client initialization
class Sovant:
    def __init__(
        self,
        api_key: Optional[str] = None,
        base_url: Optional[str] = None,
        timeout: float = 30.0,
        max_retries: int = 3,
    ): ...

# Memory operations
def memory_create(self, create: MemoryCreate) -> Dict[str, Any]: ...
def memory_get(self, id: str) -> Dict[str, Any]: ...
def memory_update(self, id: str, patch: Dict[str, Any]) -> Dict[str, Any]: ...
def memory_delete(self, id: str) -> None: ...
def memory_search(self, q: SearchQuery) -> Dict[str, Any]: ...
def memory_recall(self, query: str, limit: int = 10, thread_id: str = None) -> Dict[str, Any]: ...
def memory_create_batch(self, memories: List[Dict[str, Any]]) -> Dict[str, Any]: ...

# Thread operations
def threads_create(self, title: str, description: str = None, metadata: dict = None) -> Dict[str, Any]: ...
def threads_list(self, limit: int = 20, offset: int = 0) -> Dict[str, Any]: ...
def threads_get(self, id: str, include_memories: bool = False, limit: int = 50) -> Dict[str, Any]: ...
def threads_update(self, id: str, title: str = None, status: str = None) -> Dict[str, Any]: ...
def threads_delete(self, id: str, delete_memories: bool = False) -> None: ...

# Error class
class SovantError(Exception):
    code: str                     # Error code (e.g., 'TIMEOUT', 'RATE_LIMIT_EXCEEDED')
    status: Optional[int]         # HTTP status code (if applicable)
    details: Optional[Any]        # Additional error details