Temporal

Temporal runs long-lived, crash-proof workflows; xmemory stores durable, schema-grounded memory your agents can recall across runs, users, and services. The xmemory Temporal plugin wires the two together so a memory read or write becomes a replay-safe Temporal Activity — added to your Worker with a single line.

It ships in two languages that mirror each other:

Python — xmemory-temporal — publishing to PyPI
TypeScript — @xmemory/temporal — publishing to npm

An agent’s memory is exactly the state you don’t want to lose when a worker crashes mid-turn. Putting xmemory behind Temporal makes a memory write a durable step: it survives process death, redeploys, and rolling upgrades, and Temporal — not your code — owns its retries and timeouts.

What you get

Memory as Activities. read, write, write_async + write_status run as Activities, so all network I/O stays out of Workflow code and Workflows replay deterministically.
A durable deep write. write_durable(text) enqueues a write and polls it to completion from the Workflow, so a multi-minute extraction survives worker restarts — the poll state lives in Workflow history, not a worker process.
A near-zero-diff migration. The Workflow-side handle mirrors the plain xmemory client, so agent code that already calls inst.read(...) / inst.write(...) keeps working — it just dispatches to an Activity.
Temporal-owned retries. xmemory errors map to typed ApplicationFailures with retryable / non-retryable verdicts (see below), so you tune RetryPolicy against stable error-type strings.
Opt-in auto-capture of Activity results into memory, via an Activity interceptor that never touches the replay path.

Prerequisites

Python 3.10+ or Node.js 20+
A local Temporal dev server or a Temporal Cloud namespace
An xmemory API key and cluster

Start a local Temporal server with the Temporal CLI:

temporal server start-dev

Set your xmemory API key:

export XMEM_API_KEY="your-xmemory-api-key"

Install the plugin. Until the PyPI / npm release, install straight from the GitHub repositories:

# Python — from source (until the PyPI release)
pip install "git+https://github.com/xmemory-ai/xmemory-temporal.git"

# TypeScript — from source (until the npm release)
npm install github:xmemory-ai/xmemory-temporal-ts @temporalio/worker @temporalio/client

Once the packages are published this becomes pip install xmemory-temporal / npm install @xmemory/temporal.

Create an xmemory instance

xmemory stores information in typed instances. Generate a schema, then create an instance to hold your agent’s memory. Choose primary keys the agent will naturally restate in text — a person’s full name, or a literal customer_id / interaction_id you supply — so the diff engine can resolve repeated writes to the same record.

import json
import os

from xmemory import SchemaType, XmemoryClient

with XmemoryClient(api_key=os.environ["XMEM_API_KEY"]) as client:
    cluster_id = client.admin.list_clusters()[0].id

    schema = client.admin.generate_schema(
        cluster_id,
        (
            "Track customer-support memory. Customers are identified by customer_id. "
            "Support interactions are identified by interaction_id and record the "
            "message, recalled context, response, and outcome. Make customer_id and "
            "interaction_id primary keys so repeated writes update the same records."
        ),
    )

    instance = client.admin.create_instance(
        cluster_id=cluster_id,
        name="temporal-support-memory",
        schema_text=json.dumps(schema.data_schema),
        schema_type=SchemaType.JSON,
    )
    print(instance.id)

Run it once and save the printed instance id:

export XMEM_INSTANCE_ID="the-returned-instance-id"

You can also create and manage instances in the xmemory console. See the Python SDK guide for other schema and instance-management options.

Quickstart — Python

from temporalio.client import Client
from temporalio.worker import Worker
from xmemory_temporal import XmemoryConfig, XmemoryPlugin

from workflows import CustomerSupportWorkflow

config = XmemoryConfig(instance_id="<your-instance-id>")  # reads XMEM_API_KEY from the env
plugin = XmemoryPlugin(config)

client = await Client.connect("localhost:7233", plugins=[plugin])
# The Worker inherits the client's plugins automatically — do NOT pass it again here.
worker = Worker(client, task_queue="xmemory-support", workflows=[CustomerSupportWorkflow])
await worker.run()

Inside the Workflow, xmemory_for_workflow() returns a handle whose methods dispatch to Activities:

from temporalio import workflow

with workflow.unsafe.imports_passed_through():
    from xmemory_temporal import xmemory_for_workflow


@workflow.defn
class CustomerSupportWorkflow:
    @workflow.run
    async def run(self, customer_id: str, interaction_id: str, message: str) -> str:
        mem = xmemory_for_workflow()

        # A memory store has no ambient "current user" — name whom the fact is
        # about, then recall by that name (or pass scope= to bind to a record).
        context = await mem.read(f"What should support know about {customer_id}?")

        # ... handle the request (put any LLM call in its own Activity) ...
        response = f"Interaction {interaction_id} recorded for {customer_id}."

        # A durable write: enqueue + poll to completion, surviving worker restarts.
        await mem.write_durable(
            f"interaction_id: {interaction_id}. customer_id: {customer_id}. "
            f"message: {message}. response: {response}. status: completed."
        )
        return str(context.reader_result)  # reader_result is Any

Quickstart — TypeScript

import { NativeConnection, Worker } from '@temporalio/worker';
import { XmemoryPlugin } from '@xmemory/temporal';

const plugin = new XmemoryPlugin({ instanceId: '<your-instance-id>' }); // reads XMEM_API_KEY
const connection = await NativeConnection.connect({ address: 'localhost:7233' });
const worker = await Worker.create({
  connection,
  taskQueue: 'xmemory-support',
  workflowsPath: require.resolve('./workflows'),
  plugins: [plugin],
});
await worker.run();

import { xmemoryForWorkflow } from '@xmemory/temporal';

export async function customerSupportWorkflow(
  customerId: string,
  interactionId: string,
  message: string,
): Promise<unknown> {
  const mem = xmemoryForWorkflow();

  const context = await mem.read(`What should support know about ${customerId}?`);
  const response = `Interaction ${interactionId} recorded for ${customerId}.`;

  await mem.writeDurable(
    `interaction_id: ${interactionId}. customer_id: ${customerId}. ` +
      `message: ${message}. response: ${response}. status: completed.`,
  );
  return context.readerResult;
}

Importing xmemoryForWorkflow from the package root inside Workflow code is safe: the package is marked side-effect-free, so Temporal’s Workflow bundler tree-shakes the plugin and the xmemory client (non-Workflow-safe modules) out of the sandbox bundle.

Register the plugin once, never twice. In Python, put it on the Client (the Worker inherits it) — passing it to both registers the Activities twice and fails with “More than one activity named xmemory_read”. In TypeScript, put it on the Worker. Use one plugin instance per Worker.

Durable writes

A deep xmemory extraction can take minutes. write_durable enqueues the write and then polls its status from the Workflow, so the wait is a Temporal timer in server-side history rather than a blocked activity slot:

status = await mem.write_durable(text, max_wait=timedelta(minutes=15))

const status = await mem.writeDurable(text, { maxWaitMs: 15 * 60_000 });

Because the poll loop lives in Workflow history, you can redeploy the worker fleet mid-write and nothing is lost — the loop resumes on the new worker and completes. Under the time-skipping test environment a 15-minute write runs in milliseconds, because Temporal fast-forwards the timers. For the fire-and-forget pattern (kick off several writes, keep talking, join before the turn ends), write_async_start() and write_status() are public too.

Replay safety and idempotency

Two things keep memory operations correct under retries and replay:

Replay never re-issues an operation. All I/O is in Activities; Workflow code only schedules Activities and sleeps. Temporal replays Workflow code but never re-runs a completed Activity, so a replay never repeats a memory read or write. The test suite proves this with a forced-replay (max_cached_workflows=0) side-effects test.
Writes default to at-most-once. It is tempting to lean on xmemory’s primary-key dedup to make retries safe — a re-write of the same fact should update the same record. But primary-key extraction is non-deterministic: xmemory authors keys with a model that can normalize the same value differently across runs (e.g. Dr. Robert Kim vs Robert Kim), and a disagreement forks the entity into a new row. So a lost-response retry can duplicate. Rather than risk that silently, write Activities default to maximum_attempts=1: a failed write surfaces to your Workflow, which decides to retry, compensate, or fail. Reads and status-polls (idempotent) retry generously.

Opt into write retries only when your primary keys are literal identifiers present verbatim in the text — a customer_id / interaction_id you supply, which re-extract deterministically. Then a retry is a safe no-op update:

mem = xmemory_for_workflow(write_retry_policy=RetryPolicy(maximum_attempts=3))

const mem = xmemoryForWorkflow({ writeRetryPolicy: { maximumAttempts: 3 } });

This is why the support example above builds its write text from literal customer_id / interaction_id values — with those as schema primary keys, opting into retries is safe.

Error handling

xmemory errors become ApplicationFailures with stable type strings you can match in a RetryPolicy (non_retryable_error_types=[...] / nonRetryableErrorTypes). The mapping is derived from the server’s own error codes:

xmemory condition	`type`	Retryable?
transport error / timeout / HTTP ≥ 500 / 408	`XmemoryServerError` / `XmemoryUnavailable`	yes
`RATE_LIMITED` (429)	`XmemoryRateLimited`	yes — honors `Retry-After`
`QUOTA_EXCEEDED` + daily	`XmemoryDailyQuotaExceeded`	yes (long backoff)
`QUOTA_EXCEEDED` + monthly	`XmemoryMonthlyQuotaExceeded`	no
`QUOTA_EXCEEDED` (kind unknown)	`XmemoryQuotaExceeded`	no
`UNAUTHORIZED` / `FORBIDDEN`	`XmemoryAuthFailed`	no
`NOT_FOUND`	`XmemoryNotFound`	no
validation / conflict / schema-evolution rejections	`XmemoryBadRequest` / `XmemorySchemaRejected`	no
Activities registered without the plugin	`XmemoryNotBound`	no
an unrecognized code	`XmemoryUnknown`	yes (never fatal)

Plus three raised by the durable-write loop from a polled write_status, all non-retryable: XmemoryWriteFailed (the queued write reported failed), XmemoryWriteNotFound (the write id was not found past a grace window), and XmemoryWriteTimeout (polling exceeded max_wait).

An unrecognized error code stays retryable and never raises — a stricter client that crashed on a newer server’s code would break during rolling deploys.

Credentials and privacy

The config holds the name of the environment variable that supplies the API key (XMEM_API_KEY by default), never the key itself — so nothing secret is serialized into Activity arguments, which Temporal persists in the clear. Pass the key in-process instead with XmemoryPlugin(config, api_key=...) / new XmemoryPlugin(config, { apiKey }) if you prefer.

Your memory text and queries, however, are in history. The query you read and the text you write are Activity inputs, and the error mapping keeps raw transport strings out of failure messages — but the inputs themselves, and the reader_result, are persisted to cleartext Temporal history and shown in the Web UI. If your memory text is sensitive, install a Temporal Payload Codec to encrypt payloads at the edge — the plugin deliberately does not impose one, since a codec applies namespace-wide to every payload, not just xmemory’s.

Auto-capture (opt-in)

Instead of writing memory explicitly, you can capture the results of your own Activities into memory automatically:

from xmemory_temporal import AutoCaptureConfig, XmemoryPlugin

plugin = XmemoryPlugin(
    config,
    auto_capture=AutoCaptureConfig(
        project=lambda activity_name, result: summarize(result),  # return None to skip
        sample_rate=0.25,
    ),
)

Off by default. It runs as an Activity interceptor (outside the replay path), requires a project function that decides what — if anything — to remember, samples to bound fan-out, and never fails the wrapped Activity if a capture write errors. Capture is an enqueue (write_async) bounded by a short timeout, so it can never slow the wrapped Activity past its start_to_close. It skips any Activity whose name starts with xmemory_, and never captures Queries.

Production checklist

Register the plugin once (Python: on the Client; TypeScript: on the Worker), one instance per Worker.
Prefer write_durable when a later step needs the write to land; use read for recall.
Keep write retries off unless your primary keys are literal identifiers present verbatim in the text.
Match RetryPolicy(non_retryable_error_types=...) to the typed error strings above.
Return only the minimum recalled context — Activity results are recorded in Workflow history.
Install a Payload Codec if memory text is sensitive.
Keep Temporal and xmemory credentials in a secret manager, never in Workflow arguments.

The plugin source lives on GitHub — xmemory-temporal (Python) and xmemory-temporal-ts (TypeScript). For the underlying client APIs, see the Python and TypeScript SDK guides. For the raw tool surface, see the MCP guide.