Python

The xmemory-ai package gives your Python code persistent, structured memory. Write free-form text, have it automatically extracted into typed objects, and query it back in natural language.

For MCP-based integration (no SDK needed), see the MCP guide.

API key: To use xmemory APIs or integrations, you need an API key. Get one from the xmemory console. Copy and securely store the key. Never share your API key publicly.

The full method, data-type, enum, and exception listing is in the Reference at the bottom of this page.

---

Installation

pip install xmemory-ai

Requires Python 3.10+ and pydantic>=2.0.

---

Quick start

This is the full flow — from schema to stored knowledge to answers — in a single script:

import yaml
from xmemory import XmemoryClient, SchemaType

# Connect (reads XMEM_API_KEY from env if api_key is not passed)
client = XmemoryClient(api_key="your-api-key")

# List available clusters
clusters = client.admin.list_clusters()
cluster_id = clusters[0].id

# Describe what you want to remember
schema = client.admin.generate_schema(
    cluster_id,
    "Track contacts with name, email, company, and notes.",
)

# Create a memory instance from that schema
inst = client.admin.create_instance(
    cluster_id=cluster_id,
    name="contacts",
    schema_text=yaml.dump(schema.data_schema),
    schema_type=SchemaType.YML,
)
# inst is an InstanceAPI handle — all subsequent calls go through it.

# Write some information
inst.write("Alice Johnson works at Acme Corp. Her email is alice@acme.com.")
inst.write("Bob Lee is a designer at Globex. He joined last Monday.")

# Read it back
result = inst.read("What is Alice's email?")
print(result.reader_result)  # {"answer": "alice@acme.com"}

result = inst.read("Who joined recently?")
print(result.reader_result)  # {"answer": "Bob Lee joined last Monday."}

client.close()

Once you have an instance ID, skip the schema step on subsequent runs:

client = XmemoryClient(api_key="your-api-key")
inst = client.instance("your-saved-instance-id")

---

Configuration

Parameter	Env var	Default	Description
api_key	XMEM_API_KEY	None	API key for authentication
url	XMEM_API_URL	https://api.xmemory.ai	API base URL
timeout	—	60	Default request timeout in seconds
http_client	—	None	External httpx.Client (you manage its lifecycle)

All parameters are keyword-only. The API key and URL fall back to their environment variables when not passed.

Context manager

with XmemoryClient(api_key="your-api-key") as client:
    inst = client.instance("your-instance-id")
    result = inst.read("What is Alice's email?")

---

Async client

AsyncXmemoryClient exposes the same methods with the same signatures as the sync client — three things differ, and that’s the whole story:

• await every call — await inst.read(...), await client.admin.create_instance(...).
• aclose() instead of close() — or just use the context manager.
• async with instead of with.

from xmemory import AsyncXmemoryClient

async def main():
    async with AsyncXmemoryClient(api_key="your-api-key") as client:
        inst = client.instance("your-instance-id")
        result = await inst.read("What is Alice's email?")
        print(result.reader_result)
    # exiting `async with` calls aclose() for you (or: await client.aclose())

import asyncio
asyncio.run(main())

---

Writing

Send free-form text — xmemory extracts structured objects according to your schema and merges them into the knowledge graph.

inst = client.instance("your-instance-id")

resp = inst.write("Carol is a senior engineer at Initech. Her email is carol@initech.com.")
print(resp.changes)   # what the write created / updated / deleted
print(resp.trace_id)  # request trace id when available

WriteResult, ReadResult, and ExtractResult include trace_id, which is useful when you want to correlate SDK calls with API logs.

Extraction logic

Control the speed/accuracy tradeoff — pass extraction_logic=ExtractionLogic.DEEP to override the FAST default:

Value	When to use
FAST	High-volume, low-stakes writes (default)
DEEP	Important or complex information

Async writes

For latency-sensitive code, enqueue a write and return immediately:

resp = inst.write_async("Dave manages the London office.")
print(resp.write_id)  # use this to poll for completion

Then check the status:

status = inst.write_status(resp.write_id)
print(status.write_status)
# → WriteQueueStatus.QUEUED | PROCESSING | EXTRACTING | EXTRACTED | APPLYING | COMPLETED | FAILED | NOT_FOUND

Do not call read immediately after write_async — the data may not be committed yet. Poll with write_status until COMPLETED, or use write (synchronous) when you need to read right after.

Structured writes

When you already know exactly what to store, skip extraction: pass structured_mutations instead of text (exactly one of the two — the client raises ValueError otherwise). Each mutation is a deterministic, LLM-free create / update / delete of an object or relation, applied in list order — later mutations may reference objects created earlier in the same batch.

from xmemory import ObjectCreate, ObjectMutation, ObjectUpdate, RelationMutation

resp = inst.write(structured_mutations=[
    ObjectMutation(
        object_type="person",
        create=ObjectCreate(key={"email": "carol@initech.com"}, values={"name": "Carol"}),
    ),
    RelationMutation(
        relation_type="works_at",
        create={"endpoints": [
            {"object_name": "person", "key": {"email": "carol@initech.com"}},
            {"object_name": "company", "key": {"name": "Initech"}},
        ]},
    ),
])
print(resp.changes)  # exactly what was created / updated / deleted

Each mutation carries exactly one of create / update / delete. key identifies an object by its user-defined primary-key fields, or (for update/delete) by {"xuid": ...}. A None value in values clears that field:

inst.write(structured_mutations=[
    ObjectMutation(
        object_type="person",
        update=ObjectUpdate(key={"email": "carol@initech.com"}, values={"role": None}),
    ),
])

Plain dicts in the API wire form work anywhere a model does — handy for payloads produced by an LLM:

inst.write(structured_mutations=[
    {"object_mutation": {"object_type": "person", "delete": {"key": {"email": "carol@initech.com"}}}},
])

write_async accepts structured_mutations identically — poll with write_status as usual. See the API reference for the full wire format, relation addressing (endpoints, allow_bulk_delete), and in-batch ordering rules.

---

Reading

Ask questions in natural language. xmemory translates them into SQL against the knowledge graph and returns a formatted answer.

resp = inst.read("Who works at Acme Corp?")
print(resp.reader_result)
print(resp.trace_id)

Read modes

from xmemory import ReadMode

# Plain-text answer (default)
resp = inst.read("What is Alice's email?", read_mode=ReadMode.SINGLE_ANSWER)
# → {"answer": "alice@acme.com"}

# Structured objects and relations
resp = inst.read("Show all contacts", read_mode=ReadMode.XRESPONSE)
# → {"objects": [...], "relations": [...]}

# Raw SQL result sets
resp = inst.read("List all contacts", read_mode=ReadMode.RAW_TABLES)
# → {"tables": [...]}

Scoped reads

Restrict a query to a specific set of objects. Useful for multi-tenant systems, privacy controls, and compliance requirements.

from xmemory import ReadScope, ScopeObjectRef, RelationsScope

scope = ReadScope(
    objects=[
        ScopeObjectRef(type="Person", key={"name": "Alice Johnson"})
    ],
    relations_scope=RelationsScope.NO_RELATIONS,
)
resp = inst.read("What is their email?", scope=scope)
print(resp.reader_result)

Multiple objects with relations

scope = ReadScope(
    objects=[
        ScopeObjectRef(type="Person", key={"email": "alice@example.com"}),
        ScopeObjectRef(type="Person", key={"email": "bob@example.com"}),
    ],
    relations_scope=RelationsScope.ALL_RELATIONS,
)
resp = inst.read("Which projects do these people work on together?", scope=scope)

Scope options

• NO_RELATIONS (default): Only the specified objects.
• ALL_RELATIONS: Include junction tables between objects.

Error handling

from xmemory import XmemoryAPIError

try:
    resp = inst.read("...", scope=scope)
except XmemoryAPIError as e:
    if e.status == 400:
        print(f"Invalid scope: {e.message}")
    else:
        raise

---

Extracting (without writing)

Preview what xmemory would extract from a piece of text, without storing anything:

resp = inst.extract("Dave manages the London office.")
print(resp.objects_extracted)
print(resp.trace_id)

Accepts the same extraction_logic parameter as write.

---

Describing (agent tool discovery)

The describe() method returns agent-facing tool descriptions enriched with the instance’s actual schema. Use it to tell an LLM what tools are available and how to call them.

desc = inst.describe()

# Plain text — inject into a system prompt
print(desc.as_text())

# Anthropic tool-use format
tools = desc.as_anthropic_tools()

# OpenAI function-calling format
tools = desc.as_openai_tools()

Results are cached locally for 5 minutes. To force a refresh (e.g. after updating the schema):

inst.clear_describe_cache()
desc = inst.describe()

as_text() shows tools as method signatures by default. Pass include_http=True to also show HTTP method and path for raw REST callers.

---

Cluster and instance management

These methods live on client.admin and cover the control plane: discovering clusters, generating schemas, creating instances, and evolving them safely.

Clusters

clusters = client.admin.list_clusters()
for c in clusters:
    print(f"{c.id}: {c.name}")

Schema generation

Describe what you want to track in plain language:

schema = client.admin.generate_schema(
    cluster_id,
    "Track user preferences, open tasks with priorities, and conversation history.",
)
print(schema.data_schema)

Object names use CamelCase (UserPreferences, OpenTask). The generation endpoint handles naming conventions automatically. To evolve an existing schema (rename / remove / type change) rather than create one, use enhance_schema — it returns a structured migration plan; see Schema evolution.

Instances

Create an instance from a generated schema:

import yaml
from xmemory import SchemaType

inst = client.admin.create_instance(
    cluster_id=cluster_id,
    name="my-memory",
    schema_text=yaml.dump(schema.data_schema),
    schema_type=SchemaType.YML,
    description="Optional description",
)
# inst is a bound InstanceAPI — use inst.write(), inst.read(), etc.

For a purely additive change (new objects/fields/relations), update the schema directly — existing data is preserved:

new_schema = client.admin.generate_schema(
    cluster_id,
    "Add an assignee field to tasks.",
    current_yml_schema=current_schema,
)
client.admin.update_instance_schema(
    instance_id, yaml.dump(new_schema.data_schema), SchemaType.YML
)

List and delete instances:

instances = client.admin.list_instances()
info = client.admin.get_instance(instance_id)

client.admin.delete_instance(instance_id)

Schema evolution

Non-additive changes (rename / remove / type change) preserve data via a structured migration plan. Use enhance_schema to get the plan, dry_run_migration to preview the DDL, then update_instance_schema to apply it. confirm_destructive=True is required for ops that drop data.

current = client.admin.get_instance_schema(instance_id).data_schema

# 1. Enhance — new schema + an executor-ready migration plan.
enhanced = client.admin.enhance_schema(
    cluster_id,
    "Rename Person.mail to Person.email.",
    yaml.dump(current),
)
print(enhanced.summary)
for op in enhanced.migration_plan.ops:
    print(op)

new_yaml = yaml.dump(enhanced.data_schema)

# 2. Dry-run — preview the DDL, apply nothing.
preview = client.admin.dry_run_migration(
    instance_id, new_yaml, SchemaType.YML, migration_plan=enhanced.migration_plan,
)
print(preview.statements)

# 3. Update — apply. A rename is non-destructive, so confirm_destructive stays False.
info = client.admin.update_instance_schema(
    instance_id, new_yaml, SchemaType.YML,
    migration_plan=enhanced.migration_plan, confirm_destructive=False,
)
print(info.migration_id, info.prior_version, "->", info.new_version)

Review applied migrations:

page = client.admin.list_migrations(instance_id, limit=20)
for r in page.items:
    print(r.id, r.source, r.prior_version, "->", r.new_version)

detail = client.admin.get_migration(instance_id, page.items[0].id, include_yaml=True)
print(detail.yaml_before, detail.yaml_after)

Migration ops are exported as typed models (MigrationPlan, AddField, RenameField, RemoveObject, …). ProposalItem.op and MigrationRecord.ops stay as raw dicts for forward compatibility — use parse_migration_op(...) to validate them.

Suggestion engine (review → decide → apply)

xmemory watches read traffic and, on demand, surfaces a single rolling proposal of schema improvements. The flow is three calls — never wrap them; the checkpoints are the point.

from xmemory import DecisionInput

inst = client.instance(instance_id)

# 1. Review — proposal + optimistic-concurrency token.
review = inst.review_suggestions()
if review.status == "evolution_in_progress":
    print(f"Migration in flight; retry in {review.retry_after_seconds}s")
else:
    proposal = review.proposal
    for item in proposal.items:
        print(item.item_fingerprint, item.rationale, item.op)

    # 2. Decide — accept / reject / defer per item, in bulk.
    decided = inst.decide_suggestions(
        proposal.proposal_version,
        [DecisionInput(item_fingerprint=i.item_fingerprint, decision="accept")
         for i in proposal.items],
    )

    # 3. Apply — commit accepted decisions as one migration.
    applied = inst.apply_pending_decisions(decided.next_proposal_version)
    print(applied.status, applied.summary)

---

Error handling

All errors raise XmemoryAPIError. The exception carries an optional .status (HTTP status code), .code (structured error code, when the server returned one), and .details.

from xmemory import XmemoryAPIError, XmemoryHealthCheckError

# Check connectivity
try:
    client.check_health()
except XmemoryHealthCheckError as e:
    print(f"API unreachable: {e}")

# Handle operation errors
try:
    inst.write("...")
except XmemoryAPIError as e:
    print(f"Error (HTTP {e.status}): {e}")

XmemoryHealthCheckError is a subclass of XmemoryAPIError, so catching XmemoryAPIError covers both.

The schema-evolution methods set .code to a structured error type you can match on — e.g. stale_proposal_version, destructive_confirmation_required, non_additive_change_requires_plan:

try:
    inst.apply_pending_decisions(token)
except XmemoryAPIError as e:
    if e.code == "stale_proposal_version":
        review = inst.review_suggestions()  # re-review and retry

---

Reference

The complete method, data-type, enum, and exception listing. Methods are grouped by where they live; expand a row for its parameters, returns, and details.

Client

Constructors and client-level handles.

XmemoryClientCreate a sync client

XmemoryClient(url=None, *, timeout=60, api_key=None, http_client=None)

Parameters

urlstr

API base URL; falls back to XMEM_API_URL. The only positional arg. Defaults to None.

timeoutfloat

Default request timeout in seconds. Defaults to 60.

api_keystr

API key for authentication; falls back to XMEM_API_KEY. Defaults to None.

http_clienthttpx.Client

External HTTP client (you manage its lifecycle). Defaults to None.

AsyncXmemoryClientCreate an async client (same signature; methods are awaitable)

AsyncXmemoryClient(url=None, *, timeout=60, api_key=None, http_client=None)

Parameters

urlstr

API base URL; falls back to XMEM_API_URL. The only positional arg. Defaults to None.

timeoutfloat

Default request timeout in seconds. Defaults to 60.

api_keystr

API key for authentication; falls back to XMEM_API_KEY. Defaults to None.

http_clienthttpx.AsyncClient

External HTTP client (you manage its lifecycle). Defaults to None.

adminControl-plane accessor — clusters, schema generation, instances, schema evolution

client.admin→AdminAPI

Returns

client.admin (AsyncAdminAPI on the async client) — the receiver for the admin / control-plane methods (clusters, schema generation, instances, schema evolution).

instanceGet a handle for data operations on an instance

client.instance(instance_id)→InstanceAPI

Parameters

instance_idstrrequired

The instance ID to operate on.

Returns

A bound InstanceAPI handle (AsyncInstanceAPI on the async client) — call write / read / extract / … on it.

check_healthRaise XmemoryHealthCheckError if the API is unreachable

client.check_health()

Returns

None on success; raises XmemoryHealthCheckError otherwise.

closeClose the underlying HTTP client (no-op if you supplied your own)

client.close()

Details

Async client: use await client.aclose() instead — or prefer the context manager (with / async with).

acloseAsync counterpart of close() (AsyncXmemoryClient)

await client.aclose()

Admin methods

Control-plane methods on client.admin — clusters, schema generation, instances, and schema evolution.

list_clustersList clusters

list_clusters(*, ids=None, timeout=None)→list[ClusterInfo]

Parameters

idslist[str]

Filter to specific cluster IDs.

timeoutfloat

Per-call timeout in seconds.

get_clusterGet a cluster by ID

get_cluster(cluster_id, *, timeout=None)→ClusterInfo

Parameters

cluster_idstrrequired

The cluster ID to fetch.

timeoutfloat

Per-call timeout in seconds.

generate_schemaGenerate a schema from a plain-language description

generate_schema(cluster_id, schema_description, *, current_yml_schema=None, timeout=None)→GenerateSchemaResult

Parameters

cluster_idstrrequired

The cluster to generate the schema for.

schema_descriptionstrrequired

Plain-language description of what to track.

current_yml_schemastr

Existing schema to extend additively.

timeoutfloat

Per-call timeout in seconds.

enhance_schemaEvolve a schema → new schema + migration plan

enhance_schema(cluster_id, schema_description, current_yml_schema, *, timeout=None)→EnhanceSchemaResult

Parameters

cluster_idstrrequired

The cluster the schema belongs to.

schema_descriptionstrrequired

Plain-language description of the change.

current_yml_schemastrrequired

The current schema to evolve.

timeoutfloat

Per-call timeout in seconds.

Returns

Pass the returned migration_plan to dry_run_migration / update_instance_schema.

create_instanceCreate a new instance; returns a bound handle

create_instance(cluster_id, name, schema_text, schema_type, *, description=None, schema_description=None, timeout=None)→InstanceAPI

Parameters

cluster_idstrrequired

The cluster to create the instance in.

namestrrequired

Name for the instance.

schema_textstrrequired

The schema document text.

schema_typeSchemaTyperequired

Format of the schema (YML or JSON).

descriptionstr

Description.

schema_descriptionstr

Plain-language schema description.

timeoutfloat

Per-call timeout in seconds.

Returns

A bound InstanceAPI handle for the new instance — call inst.id to get its ID.

list_instancesList instances

list_instances(*, ids=None, timeout=None)→list[InstanceInfo]

Parameters

idslist[str]

Filter to specific instance IDs.

timeoutfloat

Per-call timeout in seconds.

get_instanceGet an instance by ID

get_instance(instance_id, *, timeout=None)→InstanceInfo

Parameters

instance_idstrrequired

The instance ID to fetch.

timeoutfloat

Per-call timeout in seconds.

get_instance_schemaGet an instance's schema

get_instance_schema(instance_id, *, timeout=None)→InstanceSchemaInfo

Parameters

instance_idstrrequired

The instance whose schema to fetch.

timeoutfloat

Per-call timeout in seconds.

update_instance_schemaUpdate a schema; pass migration_plan for non-additive changes

update_instance_schema(instance_id, schema_text, schema_type, *, migration_plan=None, confirm_destructive=False, timeout=None)→InstanceInfo

Parameters

instance_idstrrequired

The instance to update.

schema_textstrrequired

The new schema document text.

schema_typeSchemaTyperequired

Format of the schema (YML or JSON).

migration_planMigrationPlan | dict

Plan for non-additive changes (from enhance_schema).

confirm_destructivebool

Required for ops that drop data. Defaults to False.

timeoutfloat

Per-call timeout in seconds.

Returns

The migration_id / prior_version / new_version / migration_warnings fields are populated when a migration ran.

Errors

Raises XmemoryAPIError with .code non_additive_change_requires_plan (no plan) or destructive_confirmation_required (drops data without confirm_destructive=True).

update_instance_metadataUpdate instance name and description

update_instance_metadata(instance_id, name, description, *, timeout=None)→InstanceInfo

Parameters

instance_idstrrequired

The instance to update.

namestrrequired

New name for the instance.

descriptionstr | Nonerequired

New description (pass None to clear).

timeoutfloat

Per-call timeout in seconds.

delete_instanceDelete an instance

delete_instance(instance_id, *, timeout=None)→list[str]

Parameters

instance_idstrrequired

The instance ID to delete.

timeoutfloat

Per-call timeout in seconds.

Returns

The deleted instance IDs.

dry_run_migrationPreview a migration's DDL

dry_run_migration(instance_id, schema_text, schema_type, *, migration_plan=None, confirm_destructive=False, timeout=None)→DryRunResult

Parameters

instance_idstrrequired

The instance to preview against.

schema_textstrrequired

The new schema document text.

schema_typeSchemaTyperequired

Format of the schema (YML or JSON).

migration_planMigrationPlan | dict

Plan for non-additive changes.

confirm_destructivebool

Required for ops that drop data. Defaults to False.

timeoutfloat

Per-call timeout in seconds.

Returns

statements are the planned DDL; nothing is applied.

list_migrationsList applied migrations (newest first)

list_migrations(instance_id, *, limit=50, before_id=None, include_yaml=False, timeout=None)→ListMigrationsResult

Parameters

instance_idstrrequired

The instance to list migrations for.

limitint

Maximum number of records to return. Defaults to 50.

before_idstr

Page cursor: pass the previous page's next_before_id.

include_yamlbool

Include the before/after YAML in each record. Defaults to False.

timeoutfloat

Per-call timeout in seconds.

Returns

Page with next_before_id — pass it as before_id to fetch the next page.

get_migrationGet a single migration record

get_migration(instance_id, migration_id, *, include_yaml=False, timeout=None)→MigrationRecord

Parameters

instance_idstrrequired

The instance the migration belongs to.

migration_idstrrequired

The migration record ID.

include_yamlbool

Include the before/after YAML snapshots. Defaults to False.

timeoutfloat

Per-call timeout in seconds.

Returns

yaml_before / yaml_after are populated only when include_yaml=True.

Instance methods

Data-plane methods on a bound instance (client.instance(id)) — writing, reading, extracting, describing, and the suggestion engine.

writePersist data — extract objects from text, or apply structured mutations

write(text="", *, structured_mutations=None, extraction_logic=ExtractionLogic.FAST, diff_engine=None, timeout=None)→WriteResult

Parameters

textstr

Free-form text to extract from. Exactly one of text / structured_mutations must be provided (enforced with ValueError).

structured_mutationslist

Ordered list of ObjectMutation / RelationMutation models (or equivalent wire-form dicts): deterministic, LLM-free create/update/delete mutations — see Structured writes. Mutually exclusive with text.

extraction_logicExtractionLogic

Speed/accuracy tradeoff (FAST default, or DEEP). Text writes only. Defaults to ExtractionLogic.FAST.

diff_enginebool

Override the server's diff-engine setting for this write. Text writes only.

timeoutfloat

Per-call timeout in seconds.

write_asyncEnqueue a write; returns write_id immediately

write_async(text="", *, structured_mutations=None, extraction_logic=ExtractionLogic.FAST, diff_engine=None, timeout=None)→AsyncWriteResult

Parameters

textstr

Free-form text to extract from. Exactly one of text / structured_mutations must be provided.

structured_mutationslist

Same structured-mutation input as write. Mutually exclusive with text.

extraction_logicExtractionLogic

Speed/accuracy tradeoff (FAST default, or DEEP). Text writes only. Defaults to ExtractionLogic.FAST.

diff_enginebool

Override the server's diff-engine setting for this write. Text writes only.

timeoutfloat

Per-call timeout in seconds.

Returns

Poll for completion with write_status using the returned write_id.

write_statusPoll the status of an async write

write_status(write_id, *, timeout=None)→WriteStatusResult

Parameters

write_idstrrequired

The write ID returned by write_async.

timeoutfloat

Per-call timeout in seconds.

readQuery the instance in natural language

read(query, *, read_mode=ReadMode.SINGLE_ANSWER, read_id=None, timeout=None)→ReadResult

Parameters

querystrrequired

Natural-language question.

read_modeReadMode

Result shape (SINGLE_ANSWER default, XRESPONSE, or RAW_TABLES). Defaults to ReadMode.SINGLE_ANSWER.

read_idstr

Correlation ID for the read.

timeoutfloat

Per-call timeout in seconds.

Returns

reader_result is shaped by read_mode (e.g. {"answer": ...}).

extractExtract objects without writing them

extract(text, *, extraction_logic=ExtractionLogic.FAST, timeout=None)→ExtractResult

Parameters

textstrrequired

Free-form text to extract from.

extraction_logicExtractionLogic

Speed/accuracy tradeoff (FAST default, or DEEP). Defaults to ExtractionLogic.FAST.

timeoutfloat

Per-call timeout in seconds.

describeGet agent-facing tool descriptions (cached 5 min)

describe(*, timeout=None)→DescribeResult

Parameters

timeoutfloat

Per-call timeout in seconds.

clear_describe_cacheForce next describe() to fetch fresh data

clear_describe_cache()→None

get_schemaGet this instance's schema

get_schema(*, timeout=None)→InstanceSchemaInfo

Parameters

timeoutfloat

Per-call timeout in seconds.

review_suggestionsGet the rolling schema-improvement proposal

review_suggestions(*, session_id=None, timeout=None)→ReviewSuggestionsResult

Parameters

session_idstr

Session correlation ID.

timeoutfloat

Per-call timeout in seconds.

Returns

On status evolution_in_progress, back off for retry_after_seconds and retry — it's a status, not an error.

decide_suggestionsRecord accept/reject/defer decisions in bulk

decide_suggestions(proposal_version, decisions, *, session_id=None, timeout=None)→DecideSuggestionsResult

Parameters

proposal_versionstrrequired

Optimistic-concurrency token from the review.

decisionslist[DecisionInput]required

Per-item decisions — DecisionInput(item_fingerprint, decision, edits=None).

session_idstr

Session correlation ID.

timeoutfloat

Per-call timeout in seconds.

Returns

Pass next_proposal_version straight to apply_pending_decisions without re-reviewing.

Errors

Raises XmemoryAPIError with .code stale_proposal_version (token out of date) or invalid_decision_input.

apply_pending_decisionsApply accepted decisions as one migration

apply_pending_decisions(proposal_version, *, session_id=None, timeout=None)→ApplyPendingDecisionsResult

Parameters

proposal_versionstrrequired

The next_proposal_version from the decision step.

session_idstr

Session correlation ID.

timeoutfloat

Per-call timeout in seconds.

Returns

status nothing_to_apply means no accepted items were left to apply.

Errors

Raises XmemoryAPIError with .code stale_proposal_version, dependency_closure_failed, or apply_failed.

Types

Data types

ClusterInfoA cluster.

Fields

idstr

Cluster ID.

org_idstr

Owning organization ID.

namestr

Cluster name.

descriptionstr | None

Cluster description.

InstanceInfoAn instance's metadata.

Fields

idstr

Instance ID.

cluster_idstr

The cluster this instance belongs to.

namestr

Instance name.

descriptionstr | None

Instance description.

data_schemadict | None

The instance's data schema.

migration_idstr | None

Set only by update_instance_schema when a migration ran.

prior_versionint | None

Schema version before the migration (migration responses only).

new_versionint | None

Schema version after the migration (migration responses only).

migration_warningslist[str] | None

Non-fatal migration warnings (migration responses only).

InstanceSchemaInfoAn instance's schema.

Fields

data_schemadict

The instance's data schema.

WriteResultResult of a synchronous write.

Fields

write_idstr

ID of the committed write.

changesAny

What the write did, grouped into created / updated / deleted (null on responses from an older server).

trace_idstr | None

Correlate the call with API logs.

AsyncWriteResultAcknowledgement of an enqueued write.

Fields

write_idstr

Poll completion with write_status.

WriteStatusResultStatus of an async write.

Fields

write_idstr

The write being polled.

write_statusWriteQueueStatus

Current queue status.

error_detailstr | None

Failure detail when write_status is FAILED.

completed_atdatetime | None

When the write completed.

ReadResultResult of a read.

Fields

reader_resultAny

The formatted answer, shaped by read_mode (e.g. {"answer": ...}).

trace_idstr | None

Correlate the call with API logs.

ExtractResultResult of an extract (nothing persisted).

Fields

objects_extractedAny

The objects that would be stored (objects / relations).

trace_idstr | None

Correlate the call with API logs.

DescribeResultAgent-facing tool descriptions, with `as_text(include_http=False)`, `as_anthropic_tools()`, and `as_openai_tools()` format helpers.

Fields

instance_idstr

The described instance.

instance_namestr

Instance name.

schema_summarystr

Human-readable schema summary.

toolslist[ToolDescription]

One entry per agent-callable tool.

GenerateSchemaResultA generated schema.

Fields

data_schemadict

The produced schema as a dict.

EnhanceSchemaResultAn evolved schema plus an executor-ready migration plan.

Fields

data_schemadict

The produced schema as a dict.

migration_planMigrationPlan | None

Reconciled plan; pass to update_instance_schema / dry_run_migration.

summarystr | None

Human-readable summary of the change.

warningslist[dict]

Non-fatal reconciler observations.

repair_loglist[dict]

Auto-repairs applied to the LLM output.

DryRunResultPlanned migration DDL, with nothing applied.

Fields

status"ok"

Always "ok" for a successful dry run.

instance_idstr

The instance previewed against.

current_versionint

Current schema version.

statementslist[str]

The planned DDL statements.

warningslist[str]

Non-fatal warnings.

plan_summaryPlanSummary

Per-op-type breakdown.

requires_metadata_syncbool

Whether a metadata sync is needed after applying.

MigrationRecordOne applied-migration history row.

Fields

idstr

Migration record ID.

applied_atdatetime

When the migration was applied.

sourcestr

direct or suggestion_engine.

decided_bystr | None

Who decided the migration, if recorded.

prior_versionint

Schema version before.

new_versionint

Schema version after.

opslist[dict]

The raw ops applied (kept as dicts for forward-compat).

ops_summaryPlanSummary

Per-op-type breakdown.

notesstr | None

Free-text notes.

yaml_beforestr | None

Schema YAML before (only when include_yaml=True).

yaml_afterstr | None

Schema YAML after (only when include_yaml=True).

ListMigrationsResultA page of migration records, newest first.

Fields

status"ok"

Always "ok".

instance_idstr

The instance listed.

itemslist[MigrationRecord]

The records on this page.

next_before_idstr | None

Pagination cursor; None at the tail.

has_morebool

Whether more pages follow.

ReviewSuggestionsResultThe rolling schema-improvement proposal (or an in-progress status).

Fields

status"ok" | "evolution_in_progress"

evolution_in_progress means a migration is in flight.

instance_idstr

The instance reviewed.

proposalConsolidatedProposal | None

The proposal when status is "ok".

retry_after_secondsint | None

Back-off hint when evolution_in_progress.

DecideSuggestionsResultResult of recording a batch of decisions.

Fields

status"ok"

Always "ok".

instance_idstr

The instance decided on.

decisions_recordedlist[RecordedDecision]

One per recorded decision.

warningslist[DependencyWarning]

Advisory dependency warnings.

next_proposal_versionstr

Token to pass to apply_pending_decisions.

ApplyPendingDecisionsResultResult of applying accepted decisions as one migration.

Fields

status"ok" | "nothing_to_apply"

nothing_to_apply when no accepted items remained.

instance_idstr

The instance applied to.

migration_idstr | None

The committed migration's ID (when one ran).

prior_versionint

Schema version before.

new_versionint

Schema version after.

applied_itemslist[str]

Item fingerprints that were applied.

summarystr

Human-readable summary.

warningslist[str]

Non-fatal warnings.

noteslist[str]

Additional notes.

MigrationPlanAn ordered sequence of migration ops — the wire format emitted by `enhance_schema` and consumed by `update_instance_schema` / `dry_run_migration`.

Fields

opslist[MigrationOp]

The ordered ops.

MigrationOpOne migration operation — a discriminated union keyed on `op_type`.

Fields

op_typestr

Discriminator: one of add_object, remove_object, rename_object, change_object, add_field, remove_field, rename_field, change_field, add_relation, remove_relation, rename_relation, change_relation — each variant carries its own fields.

PlanSummaryPer-op-type breakdown of a migration plan.

Fields

count_by_op_typedict[str, int]

Op count keyed by op_type.

totalint

Total number of ops.

ConsolidatedProposalThe single rolling improvement proposal for an instance.

Fields

instance_idstr

The instance the proposal is for.

proposal_versionstr

Optimistic-concurrency token for decide_suggestions / apply_pending_decisions.

schema_versionint

Schema version the proposal targets.

itemslist[ProposalItem]

The accept/reject/defer-able items.

generated_atdatetime

When the proposal was generated.

noteslist[str]

Proposal-level notes.

ProposalItemOne decide-able item inside a proposal.

Fields

item_fingerprintstr

Stable ID; pass in a DecisionInput.

opdict

The proposed op (raw dict; parse with parse_migration_op).

evidence_feedback_idslist[str]

Feedback IDs that motivated the item.

evidence_query_sampleslist[str]

Sample queries that motivated the item.

frequencyint

How often the supporting signal occurred.

depends_onlist[str]

Fingerprints this item depends on.

current_decisionstr | None

Existing decision, if any.

rationalestr

Why the item was proposed.

DecisionInputOne decision in a `decide_suggestions` batch.

Fields

item_fingerprintstr

The ProposalItem to decide on.

decisionstr

One of accept, reject, defer.

editsdict | None

Optional op override (must keep the same op_type).

RecordedDecisionA persisted decision.

Fields

item_fingerprintstr

The decided item.

decision_idstr

ID of the recorded decision.

DependencyWarningAn advisory (non-blocking) dependency warning from `decide_suggestions`.

Fields

kindstr

Warning kind.

item_fingerprintstr

The item the warning concerns.

related_fingerprintslist[str]

Related item fingerprints.

related_summarieslist[str]

Human-readable summaries of the related items.

guidancestr

What to do about it.

ToolDescriptionOne agent-callable tool in a DescribeResult.

Fields

namestr

Tool name.

descriptionstr

What the tool does.

when_to_usestr

Guidance on when to call it.

parameterslist[ToolParameterDescription]

The tool's parameters.

http_methodstr

Underlying HTTP method.

http_pathstr

Underlying HTTP path.

ToolParameterDescriptionOne parameter of a ToolDescription.

Fields

namestr

Parameter name.

typestr

Parameter type.

descriptionstr

What the parameter is.

requiredbool

Whether it must be supplied.

enumlist[str] | None

Allowed values, if constrained.

defaultstr | None

Default value, if any.

Enums

Enum	Values
SchemaType	YML, JSON
ExtractionLogic	FAST, DEEP
ReadMode	SINGLE_ANSWER, RAW_TABLES, XRESPONSE
WriteQueueStatus	QUEUED, PROCESSING, EXTRACTING, EXTRACTED, APPLYING, COMPLETED, FAILED, NOT_FOUND

Exceptions

Exception	Parent	Attributes
XmemoryAPIError	Exception	.status, .code, .details (each None when absent)
XmemoryHealthCheckError	XmemoryAPIError	.status (HTTP status code or None)