> ## Documentation Index
> Fetch the complete documentation index at: https://docs.evermind.ai/llms.txt
> Use this file to discover all available pages before exploring further.
# Python Integration Patterns
> Production patterns for integrating EverOS into Python apps with the everos-cloud SDK
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](#concurrency-with-threads)). There is no async
client. Writes are already async *server-side*: `add` returns immediately and
extraction runs in the background.
## Installation
```bash theme={null}
pip install everos-cloud
export EVEROS_API_KEY="your_api_key"
```
## Client management
Create **one** client and reuse it. It manages connection pooling internally.
```python theme={null}
from everos_cloud import EverOS
client = EverOS() # reads EVEROS_API_KEY; or EverOS(api_key="...")
```
## Basic usage
```python theme={null}
# 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).
```python theme={null}
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.
```python theme={null}
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.
```python theme={null}
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:
```python theme={null}
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
* Create one `EverOS` instance and reuse it across your app and threads.
* Connection pooling and keep-alive are handled internally.
* 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**.
* 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.
## Next steps
Import conversation history at scale.
LLM-guided search for complex queries.
This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.