Memory Service ORM¶

datarobot[application-utils] ships a lightweight async ORM over the DataRobot Agentic Memory Service. Think of it as a typed, Pydantic v2-style document store that persists sessions (documents) and events (log entries) through a bearer-auth REST API, using your existing DATAROBOT_ENDPOINT and DATAROBOT_API_TOKEN credentials.

Install¶

pip install "datarobot[application-utils]"

Quick-start¶

import asyncio
from typing import Annotated

from datarobot.application_utils.persistence import (
    DRConcurrencyField,
    DRDeduplicationKey,
    DREvent,
    DRMemorySpace,
    DRRangeKey,
    DRSession,
    DRMemoryServiceClient,
    SYSTEM_PARTICIPANT,
)


# ── 1. Define your domain models ─────────────────────────────────────────────


class ChatSession(DRSession):
    """A chat session stored in the Memory Service."""

    __description_prefix__ = "chat"  # stable prefix; part of every query

    tenant: Annotated[str, DRRangeKey]  # description segment 1 (range queries)
    topic: Annotated[str, DRRangeKey]  # description segment 2 (range queries)
    chat_id: Annotated[str, DRDeduplicationKey]  # point-lookup / idempotent create
    rev: Annotated[int, DRConcurrencyField]  # mirrors server version integer
    title: str = ""  # plain metadata (payload only)


class ChatMessage(DREvent, session=ChatSession):
    """A single message in a chat session."""

    __event_type__ = "message"

    score: float = 0.0  # extra body field; round-trips through the wire


# ── 2. Use the ORM ────────────────────────────────────────────────────────────


async def demo() -> None:
    async with DRMemoryServiceClient() as client:
        # Create or adopt an existing space (idempotent via deduplication_key)
        space = await DRMemorySpace.post(
            client,
            description="My agent memory",
            deduplication_key="my-agent-space-v1",
        )

        # Create (or adopt) a session
        session = await ChatSession.post(
            space,
            tenant="acme",
            topic="billing",
            chat_id="billing-chat-001",
            title="Billing enquiry",
        )
        # `rev` is never passed in: it mirrors the server-assigned version.
        assert session.rev == 1

        # Append events
        msg = await ChatMessage.post(
            session,
            content="Hello, I need help with my invoice.",
            emitter_type="user",
            emitter_id="aabbccddeeff001122334455",  # 24-hex ObjectId
            score=0.9,
        )
        print(msg.sequence_id, msg.created_at)

        # List recent messages
        recent = await ChatMessage.last(session, n=10)

        # Fetch by range-key prefix (all billing sessions for "acme")
        billing_sessions = await ChatSession.list(space, tenant="acme", topic="billing")

        # Update session metadata (optimistic-concurrency guard via If-Match)
        await session.patch(title="Resolved billing enquiry")

        # Patch an event (guarded by its createdAt token)
        await msg.patch(score=0.5)


asyncio.run(demo())

Environment variables¶

Both are resolved automatically; pass them as constructor arguments to override.

Core concepts¶

Memory space (DRMemorySpace)¶

A namespace that owns sessions and events. Create one per-agent deployment or application context. Idempotent via deduplication_key.

space = await DRMemorySpace.post(client, deduplication_key="my-app-v1")
space2 = await DRMemorySpace.get(client, space.id)
spaces = await DRMemorySpace.list(client, deduplication_key="my-app-v1")
await space.patch(description="Updated description")
await space.delete()

Sessions (DRSession)¶

Documents stored in a memory space. Subclass DRSession and annotate fields with ORM markers:

Declare range-key fields in query order — a list query must specify a contiguous leading prefix (see §Range-key encoding below).

Lifecycle strategies (TTL)¶

By default, every DRSession subclass sends a single soft_delete lifecycle strategy on creation, triggered by a 2-year TTL (DEFAULT_SESSION_TTL_SECONDS, 63072000 seconds) — the Memory Service’s own maximum for a TTL trigger. Sessions therefore auto-clean unless you override this.

Override __lifecycle_strategies__ to use a shorter TTL (or a different strategy):

class ChatSession(DRSession):
    __description_prefix__ = "chat"
    __lifecycle_strategies__ = [
        {"type": "soft_delete", "trigger": {"ttl": 30 * 86400}},  # 30 days
    ]
    ...

Set it to an empty list to send no lifecycle strategies at all:

class ChatSession(DRSession):
    __lifecycle_strategies__ = []

Lifecycle strategies are sent on create only; session.patch(...) never touches them. Up to five strategy objects are allowed per session.

Events (DREvent)¶

An append-only log under a session. Bind to a session type with class MyEvent(DREvent, session=MySession). All plain declared fields map to the body dict.

class ChatMessage(DREvent, session=ChatSession):
    __event_type__ = "message"  # "message" | "tool_output" | "status"
    score: float = 0.0

Declared fields round-trip through any Pydantic-expressible type — nested models, list[Model], Optional[Model], enums and dicts all serialize into the event body (and session metadata) and validate back on read. This is what lets the chat-history layer nest typed ToolCall / Reasoning models inside a single message event.

Range-key encoding¶

Range keys are encoded in the session description field using a hierarchical path scheme:

description = "//" + esc(prefix) + "/" + esc(k1) + "/" + esc(k2) + "/"

esc(v) percent-encodes % (→ %25) then / (→ %2F), so values can contain arbitrary text including slashes. The leading // and trailing / after every segment create an anchored prefix — a substring-match on the service side is equivalent to a hierarchy prefix query.

Example — two sessions stored under chat/acme:

//chat/acme/billing/    ← tenant=acme, topic=billing
//chat/acme/support/    ← tenant=acme, topic=support

Query list(tenant="acme") sends description=//chat/acme/ which matches both. Query list(tenant="acme", topic="billing") sends description=//chat/acme/billing/ which matches only the first.

A subclass’ .list always sends at least its //<prefix>/ description filter — even with no range-key arguments — so ChatSession.list(space) returns only chat-prefixed sessions, never sessions of another DRSession subclass that happens to share the same space. (Trailing-slash anchoring keeps prefixes disjoint, so //chat/ never matches a //chatx/… or //loc/… session.) This subclass isolation is what lets the chat-history layer keep its Chat sessions and its EntityLocator index sessions (prefix loc) in one space without either bleeding into the other’s .list results.

⚠️ Case-insensitive caveat — the service performs a case-insensitive substring match, so Acme and acme are treated as the same tenant. Values that differ only in case will collide.

Optimistic concurrency¶

Sessions¶

Session PATCH sends an If-Match: <version> header. If the server’s version has advanced since you last fetched the session, the service returns HTTP 409 and the ORM raises DRMemoryVersionConflictError. Resolve by re-fetching (DRSession.get) and retrying.

try:
    await session.patch(title="New title")
except DRMemoryVersionConflictError:
    session = await ChatSession.get(space, id=session.id)
    await session.patch(title="New title")

Events¶

Event PATCH uses createdAt as a query-string concurrency token instead of a header. A stale token yields HTTP 422 and DRMemoryVersionConflictError.

try:
    await msg.patch(content="Corrected text")
except DRMemoryVersionConflictError:
    # Re-list to obtain fresh tokens
    events = await ChatMessage.list(session)
    msg = next(e for e in events if e.sequence_id == msg.sequence_id)
    await msg.patch(content="Corrected text")

Batch operations¶

# Atomic batch create (up to 200 events)
msgs = await ChatMessage.post_batch(
    session,
    events=[
        {"content": "First", "emitter_type": "agent"},
        {"content": "Second", "emitter_type": "agent", "score": 0.5},
    ],
)

# Atomic batch patch (up to 200 events)
await ChatMessage.patch_batch(
    session,
    updates=[
        (msgs[0], {"score": 0.9}),
        (msgs[1], {"content": "Updated second"}),
    ],
)

Emitter participant check¶

When emitter_type="user", the emitter’s ObjectId must be in the session’s participants list. The ORM raises DRMemoryBadRequestError early (before the HTTP call) if the emitter is not a participant.

The system sentinel SYSTEM_PARTICIPANT = "000000000000000000000000" is a client-side convention for agent-owned sessions. It is a valid ObjectId accepted by the service.

Error types¶

Public API¶

from datarobot.application_utils.persistence import (
    DRMemorySpace,
    DRSession,
    DREvent,
    DRDeduplicationKey,
    DRRangeKey,
    DRConcurrencyField,
    DRMemoryServiceClient,
    SYSTEM_PARTICIPANT,
    DEFAULT_SESSION_TTL_SECONDS,
    DRMemoryServiceError,
    DRMemoryNotFoundError,
    DRMemoryBadRequestError,
    DRMemoryValidationError,
    DRMemoryConflictError,
    DRMemoryVersionConflictError,
    DRMemoryRateLimitError,
    DRMemoryUnavailableError,
)

Client and models¶

DRMemoryServiceClient¶

Async HTTP client for the DataRobot Agentic Memory Service.

Resolves DATAROBOT_ENDPOINT and DATAROBOT_API_TOKEN from the environment; constructor arguments take precedence. Owns an httpx.AsyncClient unless one is injected.

The instance itself is lightweight — the resolved base URL plus a headers dict. Applications that act on behalf of many principals (a different API token per end user) are supported by constructing one DRMemoryServiceClient per principal over a single shared http_client: the shared pool carries the connections, each instance carries only an identity.

That isolation holds only while the shared client stays identity-free: construct it with no auth (request() also sends auth=None as a guard, since identity always travels in this instance’s headers) and a cookie jar that stores nothing — httpx.AsyncClient persists Set-Cookie responses in a jar shared by every request through it, which would replay one principal’s cookies on another principal’s requests. The service authenticates by header, not cookies, so refusing cookies loses nothing (see the multi-principal example below).

Parameters

import asyncio
from datarobot.application_utils.persistence import (
    DRMemorySpace,
    DRMemoryServiceClient,
)

async def main() -> None:
    async with DRMemoryServiceClient() as client:
        space = await DRMemorySpace.post(client, description="my-space")
        print(space.id)

asyncio.run(main())

import http.cookiejar

import httpx

# App startup.  The shared pool must stay identity-free: no auth
# (identity belongs on each DRMemoryServiceClient, not the
# transport), and a cookie jar that stores nothing — httpx would
# otherwise replay one principal's Set-Cookie on another's requests.
shared_http = httpx.AsyncClient(
    timeout=30.0,
    cookies=http.cookiejar.CookieJar(
        policy=http.cookiejar.DefaultCookiePolicy(allowed_domains=[])
    ),
)

def client_for(user_token: str) -> DRMemoryServiceClient:
    # Cheap: per-request construction, no new connections.
    return DRMemoryServiceClient(
        api_token=user_token,
        http_client=shared_http,
    )

async def shutdown() -> None:
    await shared_http.aclose()  # close the pool once, at app shutdown

DRMemorySpace¶

Represents a DataRobot Agentic Memory Service memory space.

Acts as the container for sessions; every session and event call is scoped to a space. Construct via the class methods post() or get() rather than directly instantiating.

Variables

DRSession¶

Abstract base class for Memory Service ORM session models.

Do not instantiate directly; subclass and declare fields with ORM markers.

Class variables¶

__description_prefix__ : Prefix injected at the start of the encoded description. Defaults to the subclass name. Keep it short and stable; it is part of every stored description and every list-query filter.

__lifecycle_strategies__ : Lifecycle strategy objects sent on session creation. Defaults to a single soft_delete strategy with a DEFAULT_SESSION_TTL_SECONDS (2 year) TTL trigger, so sessions auto-clean unless a subclass overrides this. Override with a different strategy list to change the TTL/strategy, or set to [] to send no lifecycle strategies at all.

Read-only properties¶

id : Server-assigned session UUID.

created_at : ISO-8601 creation timestamp.

version : Server-assigned version integer (optimistic-concurrency token).

session = await ChatSession.post(
    space,
    tenant="acme",
    topic="billing",
    chat_id="chat-001",
    title="Billing enquiry",
)

# By server id
session = await ChatSession.get(space, id="uuid-string")

# By dedup key
session = await ChatSession.get(space, chat_id="billing-chat-001")

# All sessions for tenant "acme"
sessions = await ChatSession.list(space, tenant="acme")

# Scoped to a user
sessions = await ChatSession.list(space, participant=user_oid)

# Combined: user + range prefix
sessions = await ChatSession.list(
    space, participant=user_oid, tenant="acme", topic="billing"
)

await session.patch(title="New title")
await session.patch(topic="support", title="Re: billing")

DREvent¶

Abstract base class for Memory Service ORM event models.

Do not instantiate directly; subclass with session=<SessionClass>.

Parameters

msgs = await ChatMessage.post_batch(
    session=my_session,
    events=[
        {"content": "Hi", "emitter_type": "user", "emitter_id": oid},
        {"content": "Hello", "emitter_type": "agent"},
    ],
)

await ChatMessage.patch_batch(
    session=my_session,
    updates=[
        (event_a, {"score": 0.9}),
        (event_b, {"content": "Updated text"}),
    ],
)

Field markers and constants¶

DRDeduplicationKey¶

Marker: field maps to the session deduplicationKey (point-lookup primary key).

At most one field per DRSession subclass may carry this marker. Enables MySession.get(space, my_key="value") exact-match point lookups and idempotent session creation (409 → adopt the existing session).

DRRangeKey¶

Marker: field maps to a segment of the session description (range/prefix queries).

Segments are appended to description in declaration order, encoded with the //prefix/seg1/seg2/ scheme. Values must be non-empty strings.

Known limitation¶

The Memory Service description filter is case-insensitive, so values differing only in case (e.g. Foo vs foo) will collide when querying.

DRConcurrencyField¶

Marker: field is kept in sync with the server version integer.

Enables user code to inspect the current optimistic-concurrency version without calling .version directly. At most one field per DRSession subclass may carry this marker.

Whether or not this marker is present, the ORM always tracks the server version internally for If-Match concurrency control on patch().

datarobot.application_utils.persistence.markers.SYSTEM_PARTICIPANT : str = '000000000000000000000000'¶

Sentinel ObjectId for sessions that belong to no specific user. This is a client-side convention; the Memory Service treats it as any other valid 24-hex ObjectId.

datarobot.application_utils.persistence.markers.DEFAULT_SESSION_TTL_SECONDS : int = 63072000¶

Default session TTL in seconds (2 years). Also the Memory Service maximum for a TTL trigger; used to build the default soft_delete lifecycle strategy.

Errors¶

exception datarobot.application_utils.persistence.DRMemoryServiceError¶

Base exception for all Memory Service ORM errors.

Parameters

exception datarobot.application_utils.persistence.DRMemoryNotFoundError¶

Raised when the requested resource does not exist (HTTP 404).

exception datarobot.application_utils.persistence.DRMemoryBadRequestError¶

Raised on client-side validation errors from the service (HTTP 400).

Common causes: an event emitter is not a session participant, an invalid ObjectId is supplied for a participant filter.

exception datarobot.application_utils.persistence.DRMemoryValidationError¶

Raised on schema validation errors from the service (HTTP 422).

This typically means the request body or query parameters failed the service’s Pydantic validation (e.g. a field is too long, a required field is missing, or mutually exclusive parameters are both supplied).

exception datarobot.application_utils.persistence.DRMemoryConflictError¶

Raised on a deduplication conflict when creating a session or DRMemorySpace (HTTP 409).

The ORM automatically adopts the existing resource (by fetching it via existing_id) rather than propagating this exception to callers in the normal post() flow.

Parameters

exception datarobot.application_utils.persistence.DRMemoryVersionConflictError¶

Raised on an optimistic-concurrency failure.

Surfaces as HTTP 409 for session patch() (stale If-Match header) and as HTTP 422 for event patch() (stale createdAt token).

Resolution: re-read the resource to get the current version, then retry.

exception datarobot.application_utils.persistence.DRMemoryRateLimitError¶

Raised when the service rejects a request due to quota or rate limits (HTTP 429).

The Memory Service enforces per-tenant trial quotas: monthly read/write counts answer 429 with a Retry-After header (seconds until the window resets), while the storage cap answers 429 without Retry-After — storage is a level, not a windowed quota, so freeing data (or upgrading) is the remedy rather than waiting.

Resolution: when retry_after is set, wait that many seconds before retrying or propagate the value to your own HTTP response so the caller can back off correctly. When it is None, retrying later will not help by itself — inspect detail for the limit that was hit.

Parameters

exception datarobot.application_utils.persistence.DRMemoryUnavailableError¶

Raised when no HTTP response was received from the service.

Covers request timeouts and transport failures (connection refused, DNS resolution, TLS errors, protocol violations). The original httpx exception is preserved as __cause__.

status_code is always None: the failure happened before a status code existed. Catching DRMemoryServiceError therefore covers both service-side errors and an unreachable service, without importing httpx at call sites.

Running integration tests¶

Integration tests are skipped by default. To run them against a live endpoint:

export DATAROBOT_ENDPOINT="https://app.datarobot.com/api/v2"
export DATAROBOT_API_TOKEN="<your-token>"

pytest tests/application_utils/persistence/acceptance -m integration -vv