Python SDK

Latest Version: 1.1.0 New

Official Python SDK for Sovant AI with enterprise-grade reliability features.

What's New in v1.1.0

✅ Retry logic with exponential backoff - Auto-retry on rate limits, server errors, timeouts
✅ Telemetry hooks - Monitor SDK behavior with callbacks
✅ Batch operations - Create multiple memories in one request
✅ Recall alias - More intuitive semantic search method
✅ Typed errors - Better error handling with specific error codes

Installation

pip install sovant==1.1.0

Quick Start

from sovant import Sovant, MemoryCreate

# Initialize with basic config
client = Sovant(api_key='sk_live_...')

# 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']}")

Advanced Initialization

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)
    retry_delay=1.0,                # Initial retry delay in seconds (default: 1.0, exponential backoff)

    # NEW: Telemetry hooks (v1.1.0)
    on_request=lambda req: print(f"→ {req['method']} {req['url']}"),
    on_response=lambda res: print(f"← {res['status']} ({res['duration']}ms)"),
    on_error=lambda err: print(f"✗ {err.code}: {err.message}")
)

Memory Operations

Create Memory

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']
))

Get Memory

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

Update Memory

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

Delete Memory

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

Search / Semantic Search

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 New

More intuitive alias for semantic search:

from sovant import SearchQuery

recalled = client.memory_recall(SearchQuery(
    query='where am I located?',
    limit=5
))

Batch Create New

Create multiple memories in a single request (max 100):

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}: {r['error']['message']}")

Automatic Retry & Error Handling

Retry Logic New

The SDK automatically retries failed requests with exponential backoff:

# 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
    retry_delay=1.0     # Initial delay: 1s, then 2s, then 4s (exponential)
)

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

Error Types New

from sovant import Sovant, SovantError, MemoryCreate

try:
    client.memory_create(MemoryCreate(data='Test'))
except SovantError as error:
    if error.code == 'TIMEOUT':
        print('Request timed out after retries')
    elif error.code == 'NETWORK_ERROR':
        print('Network connection failed')
    elif error.code == 'MAX_RETRIES':
        print('Max retries exceeded')
    elif error.code == 'INVALID_API_KEY':
        print('Invalid API key')
    elif error.code == 'RATE_LIMIT_EXCEEDED':
        print('Rate limit hit (after retries)')
    else:
        print(f"{error.code}: {error.message}")

Telemetry & Monitoring New

Monitor all SDK activity with callback hooks:

from sovant import Sovant
from datetime import datetime

def log_request(req):
    print(f"[{datetime.now()}] → {req['method']} {req['url']}")
    if req.get('body'):
        print(f"Body: {req['body']}")

def log_response(res):
    print(f"[{res['duration']:.0f}ms] ← {res['method']} {res['url']} - {res['status']}")

    # Track slow requests
    if res['duration'] > 2000:
        print('⚠️  Slow request detected!')

def log_error(err):
    print(f"✗ {err.code}: {err.message}")

client = Sovant(
    api_key='sk_live_...',
    on_request=log_request,
    on_response=log_response,
    on_error=log_error
)

# Example: Integrate with logging library
import logging

logger = logging.getLogger('sovant')

client = Sovant(
    api_key='sk_live_...',
    on_request=lambda req: logger.info('sovant_request', extra=req),
    on_response=lambda res: logger.info('sovant_response', extra=res),
    on_error=lambda err: logger.error('sovant_error', extra={'code': err.code, 'message': err.message})
)

Complete Example

import os
from sovant import Sovant, MemoryCreate, SearchQuery

# Initialize with all features
client = Sovant(
    api_key=os.getenv('SOVANT_API_KEY'),
    max_retries=3,
    on_request=lambda req: print(f"→ {req['method']} {req['url']}"),
    on_response=lambda res: print(f"← {res['status']} ({res['duration']:.0f}ms)"),
    on_error=lambda err: print(f"✗ {err.code}")
)

def main():
    try:
        # 1. Create single memory
        memory = client.memory_create(MemoryCreate(
            data='User prefers dark mode',
            type='preference',
            tags=['ui', 'settings']
        ))
        print(f"Created: {memory['id']}")

        # 2. Batch create
        batch = client.memory_create_batch([
            {'data': 'Memory 1', 'type': 'journal'},
            {'data': 'Memory 2', 'type': 'task'},
            {'data': 'Memory 3', 'type': 'insight'}
        ])
        print(f"Batch: {batch['summary']['successful']}/3 successful")

        # 3. Search/Recall
        results = client.memory_recall(SearchQuery(
            query='what are my preferences?',
            limit=10
        ))
        print(f"Found {results['total']} relevant memories")

        # 4. Get specific memory
        fetched = client.memory_get(memory['id'])
        print(f"Fetched: {fetched['content']}")

        # 5. Update
        updated = client.memory_update(memory['id'], {
            'tags': ['ui', 'settings', 'updated']
        })
        print(f"Updated: {updated['id']}")

        # 6. Delete
        client.memory_delete(memory['id'])
        print('Deleted successfully')

    except Exception as error:
        print(f'Error: {error}')

if __name__ == '__main__':
    main()

Type Hints

from typing import Dict, List, Any, Optional, Callable
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,                        # NEW v1.1.0
        retry_delay: float = 1.0,                   # NEW v1.1.0
        on_request: Optional[Callable[[Dict], None]] = None,   # NEW v1.1.0
        on_response: Optional[Callable[[Dict], None]] = None,  # NEW v1.1.0
        on_error: Optional[Callable[[SovantError], None]] = None  # NEW v1.1.0
    ): ...

# 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, q: SearchQuery) -> Dict[str, Any]: ...  # NEW v1.1.0
def memory_create_batch(self, memories: List[Dict[str, Any]]) -> Dict[str, Any]: ...  # NEW v1.1.0

# 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

Migration from v1.0.x

v1.1.0 is 100% backward compatible with v1.0.x. All existing code works unchanged.

Before (v1.0.x):

client = Sovant(api_key='sk_...')
client.memory_create(MemoryCreate(data='Test'))

After (v1.1.0) - Same code works, OR add new features:

client = Sovant(
    api_key='sk_...',
    max_retries=3,  # NEW: optional
    on_request=lambda req: print(req)  # NEW: optional
)

# Existing methods work unchanged
client.memory_create(MemoryCreate(data='Test'))

# NEW methods available
client.memory_create_batch([...])
client.memory_recall(SearchQuery(query='...'))

Best Practices

Use environment variables for API keys

import os
client = Sovant(api_key=os.getenv('SOVANT_API_KEY'))

Enable telemetry in production

import logging
logger = logging.getLogger('sovant')

client = Sovant(
    api_key=os.getenv('SOVANT_API_KEY'),
    on_error=lambda err: logger.error(f"{err.code}: {err.message}"),
    on_response=lambda res: logger.info(f"API latency: {res['duration']}ms")
)

Use batch operations for multiple memories

# Good: Single batch request
client.memory_create_batch(memories)

# Avoid: Multiple individual requests
for mem in memories:
    client.memory_create(MemoryCreate(**mem))  # Slower

Handle errors gracefully

from sovant import SovantError

try:
    client.memory_create(MemoryCreate(data='Test'))
except SovantError as error:
    # SDK automatically retried, this is final failure
    print(f"Final error after retries: {error.code}")

Use semantic search (recall) for natural language queries

# Good: Natural language
client.memory_recall(SearchQuery(query='what are my preferences?'))

# Also works: Keyword search with tags
client.memory_search(SearchQuery(tags=['preference']))

Rate Limits

The SDK automatically retries on rate limit errors (429)
Default retry strategy: 3 retries with exponential backoff (1s → 2s → 4s)

Customize retry behavior:

client = Sovant(
    api_key='sk_...',
    max_retries=5,      # More retries
    retry_delay=2.0     # Longer initial delay
)

Support

Documentation: https://sovant.ai/docs
Migration Guide: Memory v2 Migration
PyPI Package: https://pypi.org/project/sovant/
Email: support@sovant.ai

Changelog

Current stable: v1.1.0

See PyPI release history for full version history.