Python Integration Patterns

Production-ready patterns for integrating EverOS into Python applications with the official everos-cloud SDK: client management, error handling, retries, and concurrency.

The everos-cloud 1.x SDK is synchronous. For concurrency, share one client across threads (shown below). There is no async client. Writes are already async server-side: add returns immediately and extraction runs in the background.

Installation

pip install everos-cloud
export EVEROS_API_KEY="your_api_key"

Client management

Create one client and reuse it. It manages connection pooling internally.

from everos_cloud import EverOS

client = EverOS()  # reads EVEROS_API_KEY; or EverOS(api_key="...")

Basic usage

# Add: each message names its sender; timestamp defaults to now (ms)
client.add(
    session_id="session_001",
    messages=[
        {"sender_id": "user_alice", "role": "user",
         "content": "I prefer morning meetings before 10am."},
    ],
)

# Flush: force extraction for a session when you need recall right away
client.flush("session_001")

# Search: requires exactly one of user_id / agent_id
results = client.search("meeting preferences", user_id="user_alice",
                        method="hybrid", top_k=5)
for ep in (results.episodes or []):
    print(ep.score, ep.summary)

# Get: paginated fetch by memory type
profile = client.get("profile", user_id="user_alice")
print(profile.count, "profile items")

Error handling

The wrapper raises EverOSError. HTTP failures come as EverOSAPIError, which carries .status (the HTTP code) and .body (the raw error payload). Branch on .status: 4xx are your bug (don’t retry); 429 and 5xx are transient (retry with backoff).

import logging
from everos_cloud import EverOS, EverOSError, EverOSAPIError

logger = logging.getLogger(__name__)
client = EverOS()


def store(session_id: str, sender_id: str, content: str):
    try:
        return client.add(
            session_id=session_id,
            messages=[{"sender_id": sender_id, "role": "user", "content": content}],
        )
    except EverOSAPIError as e:
        if e.status in (400, 422):
            logger.error("Bad request (won't retry): %s", e.body)  # e.g. bad timestamp
        elif e.status in (401, 403):
            logger.error("Auth/permission problem: %s", e.body)
        raise
    except EverOSError as e:
        logger.error("EverOS call failed: %s", e)
        raise

Retry with backoff

Retry only transient failures (429, 5xx); let client errors fail fast.

import time
from everos_cloud import EverOSAPIError

TRANSIENT = {429, 500, 502, 503, 504}

def with_retry(fn, *args, attempts: int = 4, base: float = 0.5, **kwargs):
    for attempt in range(attempts):
        try:
            return fn(*args, **kwargs)
        except EverOSAPIError as e:
            if e.status not in TRANSIENT or attempt == attempts - 1:
                raise
            time.sleep(base * (2 ** attempt))  # 0.5s, 1s, 2s, ...

# Usage
results = with_retry(client.search, "meeting preferences",
                     user_id="user_alice", method="hybrid", top_k=5)

Concurrency with threads

The SDK is synchronous, but the client is safe to share across threads. Use a ThreadPoolExecutor to parallelize independent calls, for example searching several users at once.

from concurrent.futures import ThreadPoolExecutor, as_completed

def search_user(user_id: str, query: str):
    return user_id, client.search(query, user_id=user_id, method="hybrid", top_k=5)

user_ids = ["user_alice", "user_bob", "user_carol"]
with ThreadPoolExecutor(max_workers=8) as pool:
    futures = [pool.submit(search_user, uid, "project status") for uid in user_ids]
    for fut in as_completed(futures):
        uid, results = fut.result()
        print(uid, "->", len(results.episodes or []), "episodes")

Because writes are async server-side, you often don’t need client-side concurrency for add at all. Fire the calls and let extraction happen in the background. Reserve the thread pool for read-heavy fan-out (many searches).

Logging and monitoring

Wrap SDK calls to record latency and success:

import time, logging
from functools import wraps

logger = logging.getLogger("everos")

def timed(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        try:
            result = func(*args, **kwargs)
            logger.info("%s ok in %.0fms", func.__name__,
                        (time.perf_counter() - start) * 1000)
            return result
        except Exception as e:
            logger.error("%s failed in %.0fms: %s", func.__name__,
                         (time.perf_counter() - start) * 1000, e)
            raise
    return wrapper

@timed
def search_memory(user_id: str, query: str):
    return client.search(query, user_id=user_id, method="hybrid", top_k=5)

Best practices

Client management

Create one EverOS instance and reuse it across your app and threads.
Connection pooling and keep-alive are handled internally.

Error handling

Catch EverOSAPIError and branch on .status; catch EverOSError as the catch-all.
Retry only 429 / 5xx with exponential backoff; never retry 4xx.
The most common 422 is a timestamp in seconds. It must be unix ms.

Concurrency

Share one client across a ThreadPoolExecutor for read fan-out.
Writes are async server-side; don’t over-parallelize add.

method="hybrid" is the default; keyword for exact terms, vector for paraphrase, agentic for complex multi-part queries (use a longer timeout).
Pass include_profile=True when you also want consolidated profile items.

Getting Started

Use Cases

Integration & Patterns

Python Integration Patterns

Installation

Client management

Basic usage

Error handling

Retry with backoff

Concurrency with threads

Logging and monitoring

Best practices

Next steps

Batch Processing

Agentic Retrieval

​Installation

​Client management

​Basic usage

​Error handling

​Retry with backoff

​Concurrency with threads

​Logging and monitoring

​Best practices

​Next steps

Batch Processing

Agentic Retrieval

Installation

Client management

Basic usage

Error handling

Retry with backoff

Concurrency with threads

Logging and monitoring

Best practices

Next steps