Skip to content
mempill

Python (Advanced)

This guide covers the full Python API beyond the Python quickstart. Every symbol shown here is sourced from mempill/mempill-python/python/mempill/__init__.py, _mempill.pyi, and types.py.

Installation

Section titled “Installation”

The mempill wheel is published on PyPI and requires Python 3.11 or later:

Terminal window
pip install mempill

Contributors building from source (requires a Rust stable toolchain):

Terminal window
cd mempill/mempill-python
pip install maturin==1.14.1
maturin develop --release   # editable dev install

Constructors

Section titled “Constructors”
Opening an engine
import mempill


# In-memory — ephemeral; for tests or MCP sessions.
engine = mempill.open_in_memory()


# File-backed SQLite — durable across restarts.
engine = mempill.open("/path/to/agent.db")


# In-memory with a Python oracle (for conflict resolution).
engine = mempill.open_oracle_in_memory(oracle_object)


# File-backed with a Python oracle.
engine = mempill.open_oracle("/path/to/agent.db", oracle_object)

open and open_in_memory return Engine (an alias for PyEngine). open_oracle and open_oracle_in_memory return OracleEngine (an alias for PyOracleEngine). Both expose the same ingest_claim, query_memory, reconcile, query_audit methods, plus list_pending_adjudications, submit_adjudication, and sweep_expired_adjudications on OracleEngine. See Writing an Oracle.

Ingesting a claim — full request shape

Section titled “Ingesting a claim — full request shape”

All methods accept and return plain Python dicts. The IngestClaimRequest TypedDict (from mempill.types) documents the shape; the engine validates the dict at the Rust boundary.

ingest_claim — all fields
from mempill import ProvenanceLabel, Disposition


resp = engine.ingest_claim({
    "agent_id": "my-agent",      # str — agent identity; single-writer guarantee
    "subject":  "acme:ceo",      # str — the entity the claim is about
    "predicate": "held_by",      # str — the property being asserted
    "value": "Alice",            # any JSON-serialisable type


    # Provenance — use ProvenanceLabel helpers:
    "provenance": ProvenanceLabel.external_first_hand(),
    # or as a raw dict:
    # "provenance": {"type": "External", "kind": "ExternalFirstHand"},


    # Cardinality: "Functional" | "SetValued" | "Unknown"
    "cardinality": "Functional",


    # valid_time: real-world interval. Include valid_time_confidence inside this dict.
    # Omit the key entirely (or set None) for "I don't know the validity window".
    "valid_time": {
        "start": "2020-01-01T00:00:00Z",   # ISO-8601 UTC
        # "end": "2023-03-14T23:59:59Z",   # omit for open-ended
        "valid_time_confidence": 0.95,      # confidence in the window itself
    },


    "confidence": {
        "value_confidence": 0.95,      # [0.0, 1.0] — confidence in the value
        "valid_time_confidence": 0.95, # [0.0, 1.0] — temporal confidence
    },


    # Criticality: "Low" | "Medium" | "High" | "Critical"
    "criticality": "Medium",


    # derived_from: list of source claim UUIDs (for ModelDerived lineage)
    "derived_from": [],
})


print(resp["claim_ref"])     # UUID string
print(resp["disposition"])   # e.g. "CommittedCheap"
print(resp["contested_with"])  # [] unless Contested or PendingConflict


# Compare with Disposition enum (same string values):
assert resp["disposition"] == Disposition.CommittedCheap

Provenance label helpers

Section titled “Provenance label helpers”

ProvenanceLabel in mempill.types is a factory class — not an enum — that returns wire-shape dicts:

from mempill import ProvenanceLabel


# External, first-hand human assertion — cheap-path eligible.
ProvenanceLabel.external_user_asserted()
# → {"type": "External", "kind": "UserAsserted"}


# Tool result, sensor, system-of-record — cheap-path eligible.
ProvenanceLabel.external_first_hand()
# → {"type": "External", "kind": "ExternalFirstHand"}


# Re-ingesting content the engine previously served back. Amplification Guard (C6) watches this.
ProvenanceLabel.recall_re_entry()
# → {"type": "RecallReEntry"}


# Model-emitted / inferred content. Default for any model output.
ProvenanceLabel.model_derived()
# → {"type": "ModelDerived"}

ModelDerived claims are committed down-weighted and cannot overturn External claims. Always mark model-generated content ModelDerived.

Disposition enum

Section titled “Disposition enum”

Disposition is a str enum whose values match the Rust serde serialisation:

from mempill import Disposition


# Direct string comparison:
if resp["disposition"] == Disposition.CommittedCheap:
    ...


# Or just compare strings (same thing):
if resp["disposition"] == "CommittedCheap":
    ...

Key variants after ingest:

Value Meaning
"CommittedCheap" Fast-path commit — external provenance, no conflict
"Contested" Conflict; both values preserved; no winner selected
"QueuedForAdjudication" Oracle wired — conflict sent to oracle queue
"Superseded" Old claim superseded by valid-time succession
"Quarantined" Amplification Guard blocked a RecallReEntry echo

Querying memory

Section titled “Querying memory”
query_memory
result = engine.query_memory({
    "agent_id": "my-agent",
    "subject": "acme:ceo",
    "predicate": "held_by",
    # Optional: ISO-8601 UTC string for point-in-time query:
    # "as_of_tx_time": "2024-01-01T00:00:00Z",
})


belief = result["belief"]
print(belief["status"])           # e.g. "Resolved", "TimingUncertain", "Contested"


# When status is Resolved or TimingUncertain — primary carries the winner:
primary = belief.get("primary")
if primary:
    print(primary["fact"]["value"])             # the belief value
    print(primary["confidence"]["value_confidence"])
    vt = primary.get("valid_time") or {}
    print(vt.get("start"), vt.get("end"))


# When status is Contested — no winner; alternatives has all candidates:
for alt in belief.get("alternatives") or []:
    print(alt["fact"]["value"])

Belief status values

Section titled “Belief status values”
Status Meaning
"Resolved" Single authoritative belief after valid-time selection
"TimingUncertain" Only one claim, but no valid_time supplied — committed by tx-time
"Contested" Multiple claims; no winner; agent should surface this
"NoBelief" No claim found for this (subject, predicate)

Reconcile

Section titled “Reconcile”
reconcile
result = engine.reconcile({
    "agent_id": "my-agent",
    # List of [subject, predicate] pairs (as tuples or lists):
    "subject_lines": [("acme:ceo", "held_by")],
    # Empty list reconciles ALL subject lines for the agent:
    # "subject_lines": [],
})


print(result["oracle_escalations"])  # int — lines needing oracle
for claim_ref, disposition in result["outcomes"]:
    print(claim_ref, disposition)

Query audit

Section titled “Query audit”
query_audit (AuditQueryRequest)
audit = engine.query_audit({
    "agent_id": "my-agent",
    "claim_ref": None,         # None = all claims; or a specific UUID string
    "from_tx_time": None,      # None = from beginning; or ISO-8601 UTC string
    "limit": 100,
})


for entry in audit["entries"]:
    print(entry["event_kind"], entry["recorded_at"], entry["claim_ref"])

Use the AuditQueryRequest TypedDict from mempill.types for IDE hints:

from mempill.types import AuditQueryRequest


req: AuditQueryRequest = {
    "agent_id": "my-agent",
    "claim_ref": "8b1c02ad-7fcf-401f-8d0d-c5e2f025810a",  # filter by claim
    "from_tx_time": None,
    "limit": 50,
}
audit = engine.query_audit(req)

Error handling

Section titled “Error handling”

All exceptions inherit from MempillError:

from mempill import (
    MempillError,
    ValidationError,   # bad request shape or domain invariant violation
    NotFoundError,     # agent, claim, or adjudication handle not found
    ConflictError,     # write-lock contention (retry)
    StorageError,      # persistence layer failure
    ConfigError,       # invalid calibration parameter
    InternalError,     # engine invariant violated (indicates a bug)
)


try:
    resp = engine.ingest_claim({...})
except ValidationError as e:
    print(f"Bad request: {e}")
except ConflictError:
    # Write-lock contention — safe to retry
    ...
except MempillError as e:
    print(f"Engine error: {e}")

Async usage

Section titled “Async usage”

The Python engine is synchronous (PyO3 releases the GIL). Use asyncio.to_thread for non-blocking async usage from async def code:

asyncio.to_thread pattern
import asyncio
import mempill


engine = mempill.open_in_memory()


async def async_ingest():
    resp = await asyncio.to_thread(
        engine.ingest_claim,
        {
            "agent_id": "my-agent",
            "subject": "user",
            "predicate": "city",
            "value": "Berlin",
            "provenance": mempill.ProvenanceLabel.external_user_asserted(),
            "cardinality": "Functional",
            "confidence": {"value_confidence": 0.9, "valid_time_confidence": 0.0},
            "criticality": "Low",
            "derived_from": [],
        },
    )
    return resp


result = asyncio.run(async_ingest())
print(result["disposition"])  # CommittedCheap

For LangGraph, wrap the engine as a @tool that calls asyncio.to_thread internally — the demo’s mempill_langgraph/nodes.py shows the full pattern.

TypedDict helpers

Section titled “TypedDict helpers”

mempill.types exports TypedDicts for IDE completion and mypy:

from mempill.types import (
    IngestClaimRequest,   # write request
    IngestClaimResponse,  # ingest response
    QueryMemoryRequest,   # query request
    QueryMemoryResponse,  # query response
    ReconcileRequest,     # reconcile request
    ReconcileResponse,    # reconcile response
    AuditQueryRequest,    # audit request
    AuditQueryResponse,   # audit response
    ConfidenceDict,       # {"value_confidence": float, "valid_time_confidence": float}
)

These are for IDE assistance only — the engine accepts and returns plain dict objects.

Valid-time succession from Python

Section titled “Valid-time succession from Python”

To express temporal succession (CEO handoff), supply non-overlapping valid_time windows with valid_time_confidence ≥ 0.7:

# Alice: 2020 to 2023-03-14
engine.ingest_claim({
    "agent_id": "my-agent",
    "subject": "acme:ceo",
    "predicate": "held_by",
    "value": "Alice",
    "provenance": mempill.ProvenanceLabel.external_first_hand(),
    "cardinality": "Functional",
    "valid_time": {"start": "2020-01-01T00:00:00Z", "end": "2023-03-14T23:59:59Z", "valid_time_confidence": 0.95},
    "confidence": {"value_confidence": 0.95, "valid_time_confidence": 0.95},
    "criticality": "Medium",
    "derived_from": [],
})


# Bob: from 2023-03-15 onward
engine.ingest_claim({
    "agent_id": "my-agent",
    "subject": "acme:ceo",
    "predicate": "held_by",
    "value": "Bob",
    "provenance": mempill.ProvenanceLabel.external_first_hand(),
    "cardinality": "Functional",
    "valid_time": {"start": "2023-03-15T00:00:00Z", "valid_time_confidence": 0.95},
    "confidence": {"value_confidence": 0.95, "valid_time_confidence": 0.95},
    "criticality": "Medium",
    "derived_from": [],
})

Both claims commit as CommittedCheap or Superseded (not Contested) because the windows are non-overlapping and confidence is high.

Section titled “Related guides”