Async Client

pip install mengram-ai[async]

from mengram import AsyncMengram

m = AsyncMengram(api_key="om-your-key")

# Or use environment variable
m = AsyncMengram()

async with AsyncMengram() as m:
    results = await m.search("deployment")
    profile = await m.get_profile()
# Client automatically closed

import asyncio
from mengram import AsyncMengram

async def main():
    m = AsyncMengram()

    # Add memories
    result = await m.add([
        {"role": "user", "content": "Deployed on Railway with PostgreSQL"},
    ])

    # Upload a file (PDF, DOCX, TXT, MD)
    await m.add_file("meeting-notes.pdf")

    # Search
    results = await m.search("deployment")

    # Unified search
    all_results = await m.search_all("issues")

    # Profile
    profile = await m.get_profile()

    # Episodes & procedures
    events = await m.episodes(query="deployment")
    procs = await m.procedures(query="deploy")

    # Feedback
    await m.procedure_feedback(proc_id, success=True)

    # Graph & timeline
    graph = await m.graph()
    facts = await m.timeline(after="2026-01-01")

    # Agents & insights
    await m.run_agents()
    await m.reflect()
    insights = await m.insights()

    # Memory management
    await m.dedup()
    await m.merge("source", "target")
    await m.archive_fact("Entity", "old fact")

    # Triggers
    triggers = await m.get_triggers()

    # Webhooks
    await m.create_webhook(url="https://example.com/hook")
    hooks = await m.get_webhooks()

    # Jobs
    status = await m.job_status("job-abc")
    result = await m.wait_for_job("job-abc")

    # Close when done
    await m.close()

asyncio.run(main())

from mengram import AsyncMengram
from mengram.cloud.async_client import QuotaExceededError

async def safe_add(m, messages):
    try:
        return await m.add(messages)
    except QuotaExceededError as e:
        print(f"Quota hit: {e.action} {e.current}/{e.limit} ({e.plan})")
        # e.action = "add" | "search" | "add_file"
        # e.plan = "free" | "starter" | "pro"