memcell

Official Python SDK for MemCell.
Persistent, adaptive memory substrate for AI agents, workflows, and LLM pipelines.

pip install memcell
# or: uv add memcell

import asyncio
from memcell import AsyncMemCell

async def main():
    async with AsyncMemCell(api_key="mc_live_...") as memory:
        # 1. Recall relevant statements before an agent acts:
        recall = await memory.recall(
            namespace="acme/support",
            query="refund verification and escalation thresholds",
        )
        print(recall.prompt_context)
        # [directive] Require explicit customer confirmation before applying refunds over $500 (confidence: 0.94)

asyncio.run(main())

Core Primitives

1. recall — Query Memory Before Acting

Query verified directives, facts, preferences, and observations using natural language.

recall = await memory.recall(
    namespace="acme/support",
    query="refund verification and escalation thresholds",
)

# prompt_context is formatted and ready to drop directly into agent prompts:
print(recall.prompt_context)

2. remember — Save Statements

Save a verified directive, fact, preference, or observation so future agents inherit it immediately:

await memory.remember(
    namespace="acme/support",
    title="Require explicit customer confirmation before applying refunds over $500",
    type="directive",
)

3. report — Close the Feedback Loop

Memory adapts when told what happened. When an agent succeeds or fails after applying memory, report the outcome:

await memory.report(
    namespace="acme/support",
    statement_id="stmt_019a4b2c",
    outcome="worked",  # "worked" | "failed" | "avoided"
    reason="Customer verified and refund issued within guidelines",
)

Closed-Loop Execution (wrap_execution)

With wrap_execution, you can wrap any agent action or tool invocation in a self-healing, closed-loop memory:

import os
from memcell import AsyncMemCell

async def handle_refund(ticket_id: str, amount: int):
    async with AsyncMemCell(api_key=os.environ["MEMCELL_API_KEY"]) as memory:
        support = memory.scope("acme/support")

        # Wrap execution in a continuous learning loop:
        result = await support.wrap_execution(
            action="process_refund",
            query="refund verification requirements",
            fn=lambda ctx: process_refund(ticket_id, amount),
        )
        # Automatically reports "worked" on success (or "failed" on error) and updates confidence scores.
        return result.result

Resource Namespaces

The SDK provides direct, typed access to all MemCell platform resources across both synchronous (MemCell) and asynchronous (AsyncMemCell) clients:

Statements (memory.statements)

• list(namespace, ...): Paginated query with filtering by type, status, scope, q, sort.
• get(namespace, id): Retrieve an individual statement.
• create(namespace, title, type="directive", ...): Create a statement (directive, fact, preference, observation).
• update(namespace, id, ...): Update statement content, status, confidence, or metadata.
• delete(namespace, id): Delete a statement.
• star(namespace, id, starred=True): Star or unstar a statement.
• history(namespace, id): Retrieve complete version and mutation history.
• adopt(namespace, id, target_project_ids): Adopt a statement into other projects.
• promote(namespace, id, to_scope="common"): Promote a provisional statement to active.

Projects (memory.projects)

• list(...): List caller's accessible projects.
• list_for_owner(owner, ...): List projects belonging to an owner.
• get(namespace): Retrieve project details.
• create(name, ...): Create a new project.
• update(namespace, ...): Update project properties.
• delete(namespace): Delete a project.
• transfer(namespace, target_owner): Transfer project ownership.

Agents (memory.agents)

• list(namespace, ...): List agents registered in a project.
• get(namespace, agent_id): Retrieve agent details.
• create(namespace, name, ...): Register an agent.
• update(namespace, agent_id, ...): Update agent configuration.
• delete(namespace, agent_id): Deregister an agent.
• create_key(namespace, agent_id, ...): Generate a scoped agent API key.
• revoke_key(namespace, agent_id, key_id): Revoke an agent key.

Collaborators (memory.collaborators)

• list(namespace, ...): List project collaborators and pending invitations.
• invite(namespace, identifier, role="read"): Invite a user or email to collaborate.
• update_role(namespace, user_id, role): Update a collaborator's role.
• remove(namespace, user_id): Remove a collaborator.
• revoke_invitation(namespace, invitation_id): Revoke an invitation.

Organizations (memory.organizations)

• list(...), get(org_slug), create(slug, name, ...), update(org_slug, ...), delete(org_slug)
• list_members(org_slug, ...), update_member_role(org_slug, user_id, role), remove_member(org_slug, user_id)
• list_invitations(org_slug), invite_member(org_slug, email, role, ...), revoke_invitation(org_slug, invitation_id)

Usage & Quotas (memory.usage)

• get(owner, timeframe="30d"): Retrieve statement type breakdown and API quotas.

Account (memory.account)

• get(), update_profile(...)
• tokens.list(), tokens.create(name, ...), tokens.revoke(token_id)

Scoped Handle (memory.scope(namespace))

Provides convenient namespace-bound access:

• scoped.statements.*, scoped.agents.*, scoped.collaborators.*, scoped.scopes.list()

Synchronous Client (MemCell)

If you are working in synchronous scripts, Celery tasks, or standard Python workers, use the synchronous MemCell client with identical ergonomics:

from memcell import MemCell

memory = MemCell(api_key="mc_live_...")

recall = memory.recall(
    namespace="acme/support",
    query="refund verification and escalation thresholds",
)
for statement in recall.statements:
    print(f"[{statement.type}] {statement.title} (confidence: {statement.confidence})")

Configuration & Resilience

Machine-to-Machine (M2M) OAuth 2.0

For enterprise and backend microservices, authenticate using client credentials. Tokens are automatically exchanged and refreshed 60 seconds before expiration:

from memcell import AsyncMemCell

memory = AsyncMemCell(
    client_id=os.environ["MEMCELL_CLIENT_ID"],
    client_secret=os.environ["MEMCELL_CLIENT_SECRET"],
    scope="memory:read:acme/* memory:write:acme/backend",
)

Built-in 429 Resilience

MemCell automatically handles HTTP 429 rate limits with randomized jittered exponential backoff respecting the server's Retry-After header:

memory = AsyncMemCell(
    api_key="mc_live_...",
    max_retries=3,  # default: 3
    on_rate_limit_warning=lambda warning, response: print(
        f"Approaching rate limit: {warning.limit}"
    ),
)

Links

• Documentation: https://memcell.ai/docs
• Developer CLI: pip install memcell
• GitHub Repository: https://github.com/memcell-ai/sdk

Apache License 2.0 · © 2026 OpenOri