xmemory-ai 0.9.1

Python client library for the xmemory API

xmemory

Python client library for the xmemory API.

Quick start

from xmemory import XmemoryClient, SchemaType

client = XmemoryClient(api_key="xmem_...")  # or set XMEM_API_KEY env var

# Create an instance and start using it immediately
schema = """\
objects:
  person:
    fields:
      name:
        type: str
        required: true
        description: full name of the person
      role:
        type: str
        required: false
        description: job title or role
      location:
        type: str
        required: false
        description: city or location
relations: {}
"""

inst = client.admin.create_instance(
    cluster_id="<your-cluster-id>",
    name="my-memory",
    schema_text=schema,
    schema_type=SchemaType.YML,
)

inst.write("Alice is a software engineer based in Berlin.")
result = inst.read("What does Alice do?")
print(result.reader_result)

Bind to an existing instance

inst = client.instance("<your-instance-id>")
inst.write("Bob joined the team as a designer.")
result = inst.read("Who is on the team?")

Async quick start

import asyncio
from xmemory import AsyncXmemoryClient, SchemaType

async def main():
    async with AsyncXmemoryClient(api_key="xmem_...") as client:
        inst = await client.admin.create_instance(
            cluster_id="<cluster-id>",
            name="my-memory",
            schema_text=schema,  # same schema as above
            schema_type=SchemaType.YML,
        )
        await inst.write("Alice is a software engineer based in Berlin.")
        result = await inst.read("What does Alice do?")
        print(result.reader_result)

asyncio.run(main())

Configuration

Parameter	Env var	Default	Description
url	XMEM_API_URL	https://api.xmemory.ai	Base URL of the xmemory API
api_key	XMEM_API_KEY	None	API key for authentication
timeout	—	60	Default request timeout in seconds

Deprecation: The legacy term token (argument token= and env var XMEM_AUTH_TOKEN) is still accepted for backwards compatibility but prints an orange-colored deprecation notice on use. Migrate to api_key / XMEM_API_KEY. The legacy names will be removed in a future release.

Client structure

The client is organized into two namespaces:

• client.admin — cluster management, instance lifecycle, schema and metadata management
• client.instance(id) — instance-bound data operations (read, write, extract)

Admin API (client.admin)

# Clusters
clusters = client.admin.list_clusters()
cluster = client.admin.get_cluster("<cluster-id>")

# Instance lifecycle
inst = client.admin.create_instance(cluster_id, "name", schema, SchemaType.YML)
instances = client.admin.list_instances()
info = client.admin.get_instance("<instance-id>")
client.admin.delete_instance("<instance-id>")

# Schema management
schema = client.admin.get_instance_schema("<instance-id>")
client.admin.update_instance_schema("<instance-id>", new_schema, SchemaType.YML)

# Metadata management
client.admin.update_instance_metadata("<instance-id>", "new-name", "new description")

# Schema generation
result = client.admin.generate_schema(cluster_id, "People with name, role, and location.")
print(result.data_schema)

create_instance returns an InstanceAPI bound to the new instance, ready for data operations.

list_instances and get_instance return InstanceInfo metadata objects.

Instance API (client.instance(id))

inst = client.instance("<instance-id>")

# Read
result = inst.read("Who is on the team?")
print(result.reader_result)

# Write (synchronous)
result = inst.write("Bob joined the team on Monday as a designer.")
print(result.changes)  # what the write created / updated / deleted

# Write (async job)
job = inst.write_async("Bob joined the team on Monday as a designer.")
status = inst.write_status(job.write_id)

# Extract (without persisting)
result = inst.extract("Carol is a manager based in Berlin.")
print(result.objects_extracted)

Read modes

from xmemory import ReadMode

result = inst.read("Show people and companies", read_mode=ReadMode.XRESPONSE)

Scoped reads

By default a read may draw on any object in the instance. Pass a scope to restrict it to a specific set of concrete objects — each named by its type plus its user-defined primary key:

from xmemory import ReadScope, ScopeObject

result = inst.read(
    "What do these people do?",
    scope=ReadScope(
        objects=[
            ScopeObject(type="Person", key={"name": "Alice"}),
            ScopeObject(type="Person", key={"name": "Bob"}),
        ],
    ),
)

Only the listed objects are in scope. To also expose the relations among them, set relations_scope="all_relations" (the default is "no_relations"):

result = inst.read(
    "How are these people connected?",
    scope=ReadScope(
        objects=[
            ScopeObject(type="Person", key={"name": "Alice"}),
            ScopeObject(type="Company", key={"name": "Acme"}),
        ],
        relations_scope="all_relations",
    ),
)

Each ScopeObject is identified by its user-defined primary key (the same field name(s) used in your schema), not an internal id. key must contain at least one field. Scoped reads compose with read_mode.

Extraction logic

from xmemory import ExtractionLogic

result = inst.write("...", extraction_logic=ExtractionLogic.FAST)

Schema format

Schemas use the XMD (Xmemory Data Model) format with objects and relations:

objects:
  person:
    fields:
      name:
        type: str
        required: true
        description: full name of the person
      role:
        type: str
        required: false
        description: job title or role
  company:
    fields:
      name:
        type: str
        required: true
        description: company name
      industry:
        type: str
        required: false
        description: industry or sector
relations:
  employment:
    objects:
      person:
        type: person
        on_delete: cascade
      company:
        type: company
        on_delete: cascade
    description: person works at company

Field types: str, int, float, bool, date, datetime.

Schema evolution

Schemas can change after creation. xmemory supports safe, data-preserving migrations (rename / remove / type change) driven by structured migration ops, plus a suggestion engine that proposes improvements from real read traffic. Both paths are purely additive to this library — existing callers are unaffected.

See the Schema evolution section of the API reference for the conceptual model, and the Python guide for full walkthroughs.

Suggestion-engine flow (review → decide → apply)

The engine surfaces a single rolling proposal per instance. The minimum flow is three calls — review, decide (in bulk), apply:

from xmemory import DecisionInput, XmemoryClient

client = XmemoryClient(api_key="xmem_...")
inst = client.instance("<instance-id>")

# 1. Review — get the consolidated proposal + its concurrency token.
review = inst.review_suggestions()
if review.status == "evolution_in_progress":
    print(f"A migration is 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 one batch.
    decided = inst.decide_suggestions(
        proposal.proposal_version,
        [DecisionInput(item_fingerprint=item.item_fingerprint, decision="accept")
         for item 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)  # e.g. "ok" "added 1 field"

review_suggestions() returns a ReviewSuggestionsResult. When status == "evolution_in_progress", back off for retry_after_seconds and retry instead of blocking.

Direct migration flow (enhance → dry-run → update)

To drive a migration yourself (e.g. renaming a field), ask the server to enhance the current schema, preview the DDL, then apply it:

import yaml
from xmemory import XmemoryClient

client = XmemoryClient(api_key="xmem_...")
current = client.admin.get_instance_schema("<instance-id>").data_schema

# 1. Enhance — produce the new schema + an executor-ready migration plan.
enhanced = client.admin.enhance_schema(
    cluster_id="<cluster-id>",
    schema_description="Rename Person.mail to Person.email.",
    current_yml_schema=yaml.safe_dump(current),
)
print(enhanced.summary)
for op in enhanced.migration_plan.ops:
    print(op)

new_yaml = yaml.safe_dump(enhanced.data_schema)

# 2. Dry-run — preview the DDL without applying anything.
preview = client.admin.dry_run_migration(
    "<instance-id>", new_yaml, SchemaType.YML,
    migration_plan=enhanced.migration_plan,
)
print(preview.statements)

# 3. Update — apply. confirm_destructive=True is required for ops that drop data.
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)

Migration history

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

detail = client.admin.get_migration("<instance-id>", "<migration-id>", include_yaml=True)
print(detail.yaml_before, detail.yaml_after)

Migration ops are exported as typed models (MigrationPlan, MigrationOp, AddField, RenameField, RemoveObject, …). ProposalItem.op and MigrationRecord.ops are kept as raw dicts for forward compatibility — call parse_migration_op(...) / parse_migration_plan(...) to validate them into typed ops.

Runnable end-to-end examples live in examples/.

Context managers

Both clients support the context manager protocol and close the underlying HTTP connection on exit.

# sync
with XmemoryClient(api_key="xmem_...") as client:
    inst = client.instance("abc")
    inst.write("...")

# async
async with AsyncXmemoryClient(api_key="xmem_...") as client:
    inst = client.instance("abc")
    await inst.write("...")

External HTTP client

You can pass your own httpx.Client (or httpx.AsyncClient). The client will not be closed when the Xmemory client is closed, giving you full control over its lifecycle.

import httpx
from xmemory import XmemoryClient

http = httpx.Client(base_url="https://api.xmemory.ai", timeout=30)
client = XmemoryClient(http_client=http, api_key="xmem_...")

Health check

from xmemory import XmemoryHealthCheckError

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

Error handling

All errors raise XmemoryAPIError (or its subclass XmemoryHealthCheckError for connectivity failures). XmemoryAPIError carries an optional .status (HTTP status code), .code (structured error code, when the server returned one), .details (structured error payload), and .retry_after (the Retry-After response header in seconds, when the server sent one).

from xmemory import XmemoryAPIError

try:
    result = client.instance("abc").read("something")
except XmemoryAPIError as e:
    print(f"API error (HTTP {e.status}): {e}")

Branch on .code, not on the HTTP status. A single status can carry more than one meaning — 402 Payment Required is returned for both QUOTA_EXCEEDED and TRIAL_ENDED — so the structured .code is the discriminator, never the bare status. Pattern match on .code rather than parsing the message string.

Account / billing & rate-limit codes

HTTP	.code	Meaning	Retryable?
402	QUOTA_EXCEEDED	Tenant exhausted its plan's daily/monthly token quota.	No
402	TRIAL_ENDED	Trial over / subscription lapsed.	No
429	RATE_LIMITED	Genuine velocity/rate limit.	Yes — back off and retry, honoring .retry_after.

For QUOTA_EXCEEDED, details carries {"kind": "daily_quota_exceeded" | "monthly_quota_exceeded", "retry_after_seconds": int | None}, and when the window is resettable the server also sends a Retry-After header (surfaced as .retry_after). TRIAL_ENDED may carry details.kind == "trial_exceeded", or no details when raised by the paywall gate. The library never retries automatically — it only surfaces these values for you to act on.

import time

try:
    result = client.instance("abc").write("…")
except XmemoryAPIError as e:
    if e.code == "QUOTA_EXCEEDED":
        kind = (e.details or {}).get("kind")  # daily_quota_exceeded | monthly_quota_exceeded
        # Non-retryable: surface to the user; e.retry_after (seconds) hints when the window resets.
        raise
    elif e.code == "TRIAL_ENDED":
        # Non-retryable: prompt the user to upgrade / renew their subscription.
        raise
    elif e.code == "RATE_LIMITED":
        # Retryable: back off, honoring e.retry_after if set, then retry.
        time.sleep(e.retry_after or 1)
    else:
        raise

Schema-evolution codes

The schema-evolution endpoints return structured error codes you can pattern match on via .code rather than parsing the message — for example stale_proposal_version, dependency_closure_failed, destructive_confirmation_required, non_additive_change_requires_plan, stale_schema_version, migration_not_found, instance_not_initialised:

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

Response types

Method	Returns
admin.list_clusters()	list[ClusterInfo]
admin.get_cluster()	ClusterInfo
admin.create_instance()	InstanceAPI
admin.list_instances()	list[InstanceInfo]
admin.get_instance()	InstanceInfo
admin.get_instance_schema()	InstanceSchemaInfo
admin.update_instance_schema()	InstanceInfo
admin.update_instance_metadata()	InstanceInfo
admin.delete_instance()	list[str]
admin.generate_schema()	GenerateSchemaResult
admin.enhance_schema()	EnhanceSchemaResult
admin.dry_run_migration()	DryRunResult
admin.list_migrations()	ListMigrationsResult
admin.get_migration()	MigrationRecord
inst.read()	ReadResult
inst.write()	WriteResult
inst.write_async()	AsyncWriteResult
inst.write_status()	WriteStatusResult
inst.extract()	ExtractResult
inst.get_schema()	InstanceSchemaInfo
inst.describe()	DescribeResult
inst.review_suggestions()	ReviewSuggestionsResult
inst.decide_suggestions()	DecideSuggestionsResult
inst.apply_pending_decisions()	ApplyPendingDecisionsResult

Package publishing to pip

python -m pip install --upgrade build twine
python -m build

# test with test.pypi.org (separate account and API key required)
python -m twine upload --repository testpypi dist/*

# publish the real version when ready
python -m twine upload dist/*

# test the package
pip install xmemory-ai