Python (Advanced)

import { Aside } from ‘@astrojs/starlight/components’;

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

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

pip install mempill

To use the 0.4.0 API shown in this guide before it’s published, build from source instead (requires a Rust stable toolchain):

cd mempill/mempill-python
pip install maturin==1.14.1
maturin develop --release   # editable dev install, builds current main (0.4.0)

Constructors

import mempill

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

# File-backed SQLite — durable across restarts.
# The database file is derived automatically as base_dir/agent_{agent_id}.db.
engine = mempill.open_for_agent("/path/to/data", "my-agent")

# 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_for_agent("/path/to/data", "my-agent", oracle_object)

open_for_agent and open_in_memory return Engine (an alias for PyEngine). open_oracle_for_agent 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

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.

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
        # Date granularity (optional — explicit in Python/MCP; not inferred).
        # Lowercase snake_case values only:
        # "start_granularity": "month",     # "year" | "month" | "day" | "instant"
        # "end_granularity": "year",
    },

    "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

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

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

result = engine.query_memory({
    "agent_id": "my-agent",
    "subject": "acme:ceo",
    "predicate": "held_by",
    # Optional: rewind the transaction-time axis — "what did the engine know at this time?"
    # "as_of_tx_time": "2024-01-01T00:00:00Z",
    #
    # Optional: filter by the valid-time axis — "what was true in the real world on this date?"
    # "valid_at": "2021-06-01T00:00:00Z",
    #
    # The two parameters are independent. valid_at selects which claim's real-world validity
    # window covers the given instant; as_of_tx_time limits which claims the engine considers
    # based on when they were recorded.
})

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"])

    # Valid-time fields — raw datetimes and pre-rendered display strings:
    print(primary.get("valid_from_display"))   # e.g. "2020", "2020-03", "2020-03-15", or None
    print(primary.get("valid_until_display"))  # same — rendered per recorded granularity

    vt = primary.get("valid_time") or {}
    print(vt.get("start"), vt.get("end"))      # raw ISO-8601 UTC strings or None
    print(vt.get("start_granularity"))         # e.g. "year", "month", "day", "instant", or None
    print(vt.get("end_granularity"))           # same

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

Date granularity in query responses

Every query_memory response includes pre-rendered display strings alongside the raw valid-time datetimes. The display format reflects the precision that was recorded at ingest time — no fabricated day-level precision is inserted:

# Response when a Month-granular start was ingested:
{
    "belief": {
        "status": "Resolved",
        "primary": {
            "claim_ref": "8b1c02ad-...",
            "fact": {"value": "Alice"},
            "confidence": {"value_confidence": 0.9, "valid_time_confidence": 0.9},
            "valid_time": {
                "start": "2020-03-01T00:00:00Z",   # internal canonical timestamp
                "end": None,
                "valid_time_confidence": 0.9,
                "start_granularity": "month",       # lowercase: "year" | "month" | "day" | "instant"
                "end_granularity": None,
            },
            "valid_from_display":       "2020-03",  # Month granularity → no day
            "valid_until_display":      None,       # open-ended
            ...
        },
        "alternatives": []
    }
}

valid_from_display / valid_until_display (on the belief slot) are the values to show users and agents. valid_time.start_granularity / valid_time.end_granularity are the machine-readable precision labels for downstream logic.

Legacy claims (ingested before granularity was introduced) have None granularity and display in YYYY-MM-DD fallback format.

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)

Query history (raw `query_history`)

The raw engine.query_history({...}) dict path underlies the ergonomic history() helper (see the Quickstart). Since 0.4.0 each entry also carries honest per-endpoint date granularity, identical in rendering to query_memory:

result = engine.query_history({
    "agent_id": "my-agent",
    "subject": "acme:ceo",
    "predicate": "held_by",
})

for entry in result["entries"]:
    print(entry["value"], entry["status"])          # "Alice" "Superseded", etc.
    print(entry.get("valid_from_display"))           # e.g. "2020-03", "2020", None
    print(entry.get("valid_until_display"))          # DERIVED endpoint — see note below
    print(entry.get("valid_from_granularity"))       # "year" | "month" | "day" | "instant" | None
    print(entry.get("valid_until_granularity"))       # same, but see note below

valid_until_granularity is a derived value. valid_until is not stored on the entry itself — it is bounded by the successor claim’s canonical ordering key (supersession). The honest granularity to report is therefore the successor’s start_granularity, not this entry’s own (never-populated) end_granularity. When the successor’s ordering key falls back to transaction_time (low valid-time confidence), valid_until_granularity is None — a machine-assigned timestamp has no date precision to report. The last (Current, open-ended) entry always has valid_until_granularity = None. See Changelog: Date granularity on history() for the full rule.

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

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

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

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

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

mempill.types exports TypedDicts for IDE completion and mypy:

from mempill.types import (
    IngestClaimRequest,   # write request
    IngestClaimResponse,  # ingest response
    QueryMemoryRequest,   # query request (includes optional valid_at field)
    QueryMemoryResponse,  # query response
    ReconcileRequest,     # reconcile request
    ReconcileResponse,    # reconcile response
    AuditQueryRequest,    # audit request
    AuditQueryResponse,   # audit response
    ConfidenceDict,       # {"value_confidence": float, "valid_time_confidence": float}
    ValidTimeDict,        # {"start"?: str, "end"?: str, "valid_time_confidence"?: float}
    BeliefProjection,     # full belief shape returned by query_memory
    FactDict,             # {"value": any}
    BeliefSlot,           # single candidate in a belief projection
)

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

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.

Point-in-time valid-time query (`valid_at`)

After storing a succession of CEO claims (see above), query who held the role on a specific real-world date using valid_at:

# Who was CEO on 2021-06-01 in real-world time?
result = engine.query_memory({
    "agent_id": "my-agent",
    "subject": "acme:ceo",
    "predicate": "held_by",
    "valid_at": "2021-06-01T00:00:00Z",
})

belief = result["belief"]
primary = belief.get("primary")
if primary:
    print(primary["fact"]["value"])   # "Alice" — her valid-time window covers 2021-06-01

valid_at selects by the real-world validity window of each claim. It is independent of as_of_tx_time, which rewinds the transaction-time axis (when the engine recorded the claim). Both can be supplied together for full bi-temporal point-in-time queries.

Subject-scoped enumeration (`query_subject`)

query_subject returns the resolved belief for every predicate known about a subject — useful when the caller does not know the predicate names upfront:

result = engine.query_subject({
    "agent_id": "my-agent",
    "subject": "acme:ceo",
    # Optional, same semantics as query_memory:
    # "valid_at": "2021-06-01T00:00:00Z",
    # "as_of_tx_time": "2024-01-01T00:00:00Z",
})

for entry in result:
    print(entry["predicate"], entry["value"], entry["status"])
    print(entry["valid_from_display"], entry["valid_until_display"])
# held_by  Alice  Resolved
# 2020-03  None

Entries are sorted by predicate (alphabetical). Each entry mirrors the primary shape from query_memory: predicate, value, status, valid_from_display, valid_until_display, provenance, claim_ref, conf. A predicate with status: "Contested" has value: None — surface it the same way you would a Contested result from query_memory.

Quickstart (Python) — five-minute intro
Writing an Oracle — open_oracle / open_oracle_in_memory
MCP Integration — using mempill via Claude Desktop without embedding the wheel
Query Patterns — belief shapes, audit forensics