Client construction | Brain Docs

Recall() — synchronous client

stable

class Recall:
    def __init__(
        self,
        *,
        url: Optional[str] = None,
        api_key: Optional[ApiKeyResolver] = None,
        api_version: str = "2026-04-30",
        timeout: float = 60.0,
        max_retries: int = 3,
        retry_delay: float = 0.5,
        _transport: Optional[SyncHttpTransport] = None,
    ) -> None: ...

Recall wraps httpx.Client and exposes nine resources as attributes. All methods on those resources are blocking — they return decoded JSON values directly.

Constructor kwargs

OptionTypeDefault / Env

urlstr

Server base URL, e.g. 'https://api.recall.arc-labs.ai'. Falls back to the RECALL_URL or RECALL_BASE_URL environment variable. Required — constructing without one raises ValueError.

api_keystr | Callable[[], str] | Callable[[], Awaitable[str]]

Bearer token. Three accepted forms: a static string ('rcall_…'), a sync callable returning a fresh string each call (for token rotation), or an async callable returning an awaitable string (for the async transport). Falls back to RECALL_API_KEY.

api_versionstrdefault: '2026-04-30'

Sent as the Recall-Api-Version header on every request. Override only when your server pins a newer version.

timeoutfloatdefault: 60.0

Per-request timeout in seconds, applied to the underlying httpx.Client. The default of 60 seconds matches the TypeScript SDK and accommodates the longest LLM-bound writes.

max_retriesintdefault: 3

Maximum automatic retries on retriable statuses (408, 429, 500, 502, 503). Set to 0 to disable retries entirely.

retry_delayfloatdefault: 0.5

Base delay in seconds for exponential backoff. The actual sleep is retry_delay × 2^attempt × uniform(0.5, 1.5), or the server's Retry-After header when present.

_transportSyncHttpTransport

Test bypass — inject a custom transport (or a fake satisfying the SyncTransport Protocol) and the constructor will skip URL/env validation. Use this in tests; never in production.

Resources

Every instance exposes the same resources:

OptionTypeDefault / Env

memoriesSyncMemories

Write, search, get, list, create, update, delete, feedback, forget. See Memories.

entitiesSyncEntities

List, get, get_graph, merge. See Entities.

contextSyncContext

build() — assemble a prompt from the memory store. See Context.

pipelinesSyncPipelines

Non-streaming and streaming write/search variants. See Pipelines.

identitySyncIdentity

me() and refresh() — inspect the bound scope.

namespacesSyncNamespaces

Admin-plane CRUD plus reembed/reextract. JWT only.

jobsSyncJobs

get(job_id) for polling reembed/reextract progress.

observabilitySyncObservability

consistency_scan_latest() — admin-plane consistency report.

healthSyncHealth

check() — connectivity probe.

Admin plane resources

These resources require an API key with admin scope:

# Admin plane (requires API key with admin scope)
recall.agents       # SyncAgents  — create / list / get / patch / delete / stats
recall.keys         # SyncApiKeys — create / list / get / delete / rotate
recall.org          # SyncOrg     — profile / members / plan
recall.with_options(...)  # Returns new Recall with merged config

OptionTypeDefault / Env

agentsSyncAgents

Create, list, get, patch, delete, and inspect stats for agent personas. See Agents.

keysSyncApiKeys

Issue, inspect, revoke, and atomically rotate API keys. See Keys.

orgSyncOrg

Manage org profile, members, and inspect plan usage. See Org.

with_options()

Returns a new Recall (or AsyncRecall) instance with the given config overrides applied on top of the current client's settings. The new client shares no connection pool with the original — each has its own independent httpx.Client instance.

new_client = recall.with_options(
    api_key="sk_...",    # rotate to a different key
    timeout=120.0,       # extend timeout for a slow operation
)

Useful for multi-tenant code where each request should run as a different API key without constructing a brand-new client from scratch:

# Base client from env vars
base = Recall(url="https://api.recall.arc-labs.ai", api_key="rcall_default")

# Per-tenant override — only the key changes; url, retries, etc. are inherited
tenant_client = base.with_options(api_key=tenant_api_key)
result = tenant_client.memories.write(messages=[...])

Any kwarg accepted by Recall.__init__ can be passed to with_options(). Omitting a kwarg inherits the current client's value for that field.

Lifecycle

Recall is a context manager:

from recall import Recall

with Recall(url='https://api.recall.arc-labs.ai', api_key='rcall_…') as recall:
    recall.memories.write(messages=[...])
# transport closed automatically

You can also call recall.close() explicitly. After close, further requests raise from the underlying httpx client.

AsyncRecall() — asynchronous client

stable

class AsyncRecall:
    def __init__(
        self,
        *,
        url: Optional[str] = None,
        api_key: Optional[ApiKeyResolver] = None,
        api_version: str = "2026-04-30",
        timeout: float = 60.0,
        max_retries: int = 3,
        retry_delay: float = 0.5,
        _transport: Optional[AsyncHttpTransport] = None,
    ) -> None: ...

AsyncRecall wraps httpx.AsyncClient. Every resource method is a coroutine — you must await it. The kwargs are identical to Recall's.

import asyncio
from recall import AsyncRecall

async def main():
    async with AsyncRecall(
        url='https://api.recall.arc-labs.ai',
        api_key='rcall_…',
    ) as recall:
        result = await recall.memories.write(messages=[...])
        hits = await recall.memories.search(query='…')

asyncio.run(main())

async with is the recommended cleanup pattern; await recall.aclose() works for cases where the lifetime is structured but the construction site isn't an async with block.

RecallConfig

RecallConfig is the dataclass that backs both clients. You don't normally construct it directly — the client constructors create one from your kwargs. It is exposed because it documents which fields are shared between the sync and async paths.

from recall.config import RecallConfig

config = RecallConfig(
    url='https://api.recall.arc-labs.ai',
    api_key='rcall_…',
    api_version='2026-04-30',
    timeout=60.0,
    max_retries=3,
    retry_delay=0.5,
)

The config holds no scope field by design. The server derives the bound (user_id, agent_id, namespace_id) from the API key on every request — the SDK never constructs a scope.

ApiKeyResolver

ApiKeyResolver is the type alias for the three accepted shapes of api_key:

ApiKeyResolver = Union[
    str,                        # Static token
    Callable[[], str],          # Sync callable — re-invoked every request
    Callable[[], Awaitable[str]],  # Async callable — async transport only
]

The transport invokes the callable on every request, so token rotation is trivially supported:

def get_token() -> str:
    return secrets_manager.fetch('recall-api-key')

recall = Recall(url=URL, api_key=get_token)

For the async client, the callable can return an awaitable:

async def get_token() -> str:
    return await secrets_manager.fetch_async('recall-api-key')

async with AsyncRecall(url=URL, api_key=get_token) as recall:
    ...

Don't pass an async callable to the sync Recall client — the sync transport calls the resolver as a blocking function and won't await. Pass a sync wrapper (asyncio.run(...) is rarely the right answer here; cache the token in a thread-safe accessor instead).

Transport injection for tests

The _transport= kwarg accepts any object satisfying the SyncTransport or AsyncTransport Protocol — a request(method, path, body=…, params=…, idempotency_key=…) method and a stream(method, path, body=…, params=…) method. The constructor short-circuits URL validation when this kwarg is passed, so you can build a recording fixture without setting RECALL_URL.

class FakeTransport:
    def __init__(self) -> None:
        self.calls: list = []

    def request(self, method, path, *, body=None, params=None, idempotency_key=None):
        self.calls.append((method, path, body))
        return {'memoryIds': ['m_test_1']}

    def stream(self, method, path, *, body=None, params=None):
        return iter([])

fake = FakeTransport()
recall = Recall(_transport=fake)
recall.memories.write(messages=[{'role': 'user', 'content': 'hi'}])
assert fake.calls[0][1] == '/v1/remember'

Production patterns

Reuse one client per process. The underlying httpx.Client pools connections; constructing a new client per request defeats keep-alive.
Set max_retries=0 inside an outer retry budget. If your service already wraps every call in a retry loop, the SDK's own retries multiply your retry budget. Set both retry budgets explicitly.
Use RECALL_URL / RECALL_API_KEY env vars in containers. Don't bake the API key into the image.

The constructor's keyword-only-arguments (* separator) are deliberate. Positional Recall('https://…', 'rcall_…') is a type error — you must write Recall(url='https://…', api_key='rcall_…').

On this page