Python SDK

pip install mengram-ai

pip install mengram-ai[async]

from mengram import Mengram

# Pass API key directly
m = Mengram(api_key="om-your-key")

# Or use environment variable
# export MENGRAM_API_KEY=om-your-key
m = Mengram()

result = m.add([
    {"role": "user", "content": "We fixed the OOM with Redis cache"},
    {"role": "assistant", "content": "Noted the Redis cache fix."},
])
# Returns: {"status": "accepted", "job_id": "job-..."}

# With provenance tracking
m.add(messages, source="slack", metadata={"channel": "#engineering"})

# Agent mode — extract from both user and assistant messages
m.add(messages, agent_mode=True)

m.add_text("Meeting notes: decided to migrate to PostgreSQL 16")

# With provenance
m.add_text("Bug fix deployed", source="discord", metadata={"server": "dev-team"})

result = m.add_file("meeting-notes.pdf")
# {"status": "accepted", "job_id": "job-...", "file_type": "pdf", "page_count": 12, "quota_used": 12}

# Poll for completion
m.wait_for_job(result["job_id"])

results = m.search("database preferences", limit=10)
for r in results:
    print(f"{r['entity']} — score: {r['score']:.2f}")
    for fact in r.get("facts", []):
        print(f"  - {fact}")

results = m.search_all("deployment")
print(results["semantic"])     # entities
print(results["episodic"])     # events
print(results["procedural"])   # workflows

memories = m.get_all()           # list all entities
full = m.get_all_full()          # with full facts, relations, knowledge

entity = m.get("PostgreSQL")     # get specific entity
m.delete("PostgreSQL")           # delete entity

profile = m.get_profile()
print(profile["system_prompt"])   # "You are talking to Ali..."
print(profile["facts_used"])      # number of facts used

result = m.rules(format="claude_md")
print(result["content"])   # structured rules file

events = m.episodes(query="auth bug", limit=5)
recent = m.episodes(limit=20)
jan = m.episodes(after="2026-01-01", before="2026-02-01")

procs = m.procedures(query="deploy")
all_procs = m.procedures(limit=50)

m.procedure_feedback(proc_id, success=True)
m.procedure_feedback(proc_id, success=False,
    context="Build OOM", failed_at_step=3)

history = m.procedure_history(proc_id)   # all versions
evolution = m.procedure_evolution(proc_id)  # what changed and why

g = m.graph()
print(g["nodes"])  # entities
print(g["edges"])  # relationships

facts = m.timeline(after="2026-01-01", before="2026-02-01")
for f in facts:
    print(f["created_at"], f["entity"], f["fact"])

feed = m.feed(limit=20)

s = m.stats()

m.dedup()                       # find and merge duplicate entities
m.dedup_all()                   # deduplicate facts across all entities
m.dedup_entity("PostgreSQL")    # deduplicate facts on one entity
m.merge("src", "target")       # merge two entities
m.merge_user()                  # merge "User" entity into primary person
m.archive_fact("Entity", "old fact")  # archive a fact
m.fix_entity_type("React", "technology")  # fix entity type
m.reindex()                     # re-embed all entities

m.reflect()                    # trigger reflection
insights = m.insights()        # get AI insights
refs = m.reflections(scope="cross")  # scope: entity, cross, temporal

m.run_agents()                 # run all agents
m.run_agents(agent="curator")  # run specific agent
history = m.agent_history(limit=5)  # run history
status = m.agent_status()      # which agents are due

triggers = m.get_triggers()                    # get triggers
m.process_triggers()                           # fire pending triggers
m.dismiss_trigger(trigger_id)                  # dismiss without firing
m.detect_triggers(target_user_id="alice")      # detect triggers for user

m.create_webhook(
    url="https://example.com/hook",
    event_types=["memory_add", "memory_update"],
    secret="hmac-secret",
)
hooks = m.get_webhooks()
m.update_webhook(webhook_id=1, active=False)
m.delete_webhook(webhook_id=1)

team = m.create_team("Engineering", description="Shared memory")
code = team["invite_code"]      # share this code

m.join_team("abc123")           # join via invite code
teams = m.get_teams()           # list your teams
members = m.team_members(team_id=1)

m.share_memory("PostgreSQL", team_id=1)    # share with team
m.unshare_memory("PostgreSQL", team_id=1)  # make personal again
m.leave_team(team_id=1)
m.delete_team(team_id=1)        # owner only

keys = m.list_keys()
new_key = m.create_key(name="production")
m.rename_key(key_id="...", name="staging")
m.revoke_key(key_id="...")

billing = m.get_billing()             # plan, usage, quotas
checkout = m.create_checkout("pro")   # upgrade to Pro
portal = m.create_portal()           # manage subscription

status = m.job_status("job-abc123")          # check status
result = m.wait_for_job("job-abc123",        # poll until done
    poll_interval=1.0, max_wait=60.0)

# Import ChatGPT export
m.import_chatgpt("~/Downloads/chatgpt-export.zip")

# Import Obsidian vault
m.import_obsidian("~/Documents/MyVault")

# Import text/markdown files
m.import_files(["notes.md", "journal.txt"])

from mengram import Mengram
from mengram.cloud.client import QuotaExceededError

m = Mengram()

try:
    m.add([{"role": "user", "content": "Hello"}])
except QuotaExceededError as e:
    print(e.action)   # "add" or "search"
    print(e.limit)    # 50
    print(e.current)  # 50
    print(e.plan)     # "free"
    print(e)          # "Quota exceeded for 'add': 50/50 (plan: free). Upgrade at ..."

m.search("test")
print(m.quota)
# {"add": {"used": 5, "limit": 30}, "search": {"used": 12, "limit": 100}}

# Check before hitting the wall
if m.quota.get("search", {}).get("used", 0) >= m.quota.get("search", {}).get("limit", 1) * 0.8:
    print("Warning: approaching search quota limit")