memosift

Persistent memory + tool-result intelligence for AI agents — Python SDK.

MemoSift is a memory layer for AI agents that combines three things competitors don't usually combine:

1. Deterministic classification + metadata + security scanning of tool results — runs locally in the SDK, no cloud dependency
2. A queryable knowledge graph of memories, intents, entities, and artifacts — backed by Postgres + pgvector + cross-encoder reranker
3. Optional context-budget compression — auto-stub large tool outputs and auto-compress old turns when the context window fills

Same-named SDKs in Python and TypeScript, both client-side, both MIT. The cloud is a paid service.

---

Install

pip install memosift                     # core only (Mode A: local primitives)
pip install "memosift[cloud]"            # + cloud sync (Modes B / C)

# Optional framework integrations:
pip install "memosift[anthropic]"        # AsyncAnthropic adapter
pip install "memosift[openai]"           # AsyncOpenAI adapter
pip install "memosift[langchain]"        # LangChain BaseCallbackHandler
pip install "memosift[claude-code]"      # MCP server for Claude Code

Requires Python 3.12+.

---

Pro tier — getting an API key

Anything beyond Mode A (Inspector primitives — classify, extract_metadata, scan) requires a MemoSift cloud account. The asynchronous extraction + reconciliation pipeline that powers track, recall, compress, and explore is hosted at https://dev.memosift.com.

To enable the Pro features:

1. Go to memosift.com and sign up.
2. Create a project from the dashboard.
3. Create an API key inside that project (it starts with msk_).
4. Configure your SDK to point at the cloud at https://dev.memosift.com.
5. Pass the key when constructing MemoSift(...).
6. You're live — track, recall, compress, explore all work.

Wired up:

import os
from memosift import MemoSift

ms = MemoSift(
    api_key=os.environ["MEMOSIFT_API_KEY"],     # the msk_... key from step 3
    base_url="https://dev.memosift.com",        # the cloud URL from step 4
)

Or via environment so callers don't need to hard-code anything:

export MEMOSIFT_API_KEY=msk_...
export MEMOSIFT_URL=https://dev.memosift.com

ms = MemoSift(
    api_key=os.environ["MEMOSIFT_API_KEY"],
    base_url=os.environ.get("MEMOSIFT_URL", "https://dev.memosift.com"),
)

If the cloud is unreachable, every track call queues locally and ships once the network returns — your agent never blocks on MemoSift.

---

Quick start

import asyncio, os
from memosift import MemoSift

async def main():
    ms = MemoSift(
        api_key=os.environ["MEMOSIFT_API_KEY"],
        base_url=os.environ.get("MEMOSIFT_URL", "https://dev.memosift.com"),
    )

    # After each agent turn:
    await ms.track(
        [
            {"role": "user", "content": "find me CSV files"},
            {"role": "assistant", "content": "Found 3 CSVs."},
        ],
        session_id="my-session",
    )

    # When you want recall:
    result = await ms.recall(query="user's question", session_id="my-session")
    for item in result.items:
        print(item.content[:120])

    # When you want a structured context block:
    ctx = await ms.compress(session_id="my-session")
    print(ctx.context_block)

asyncio.run(main())

---

Three integration modes

Mode	Network	Mutates tool results	Provides recall	Provides compression	Use when
A — Inspector	None	No	No	No	Compliance, telemetry, routing — pure local primitives
B — Sidecar (default)	Yes	No	Yes (cross-session)	No	Long-running agents that need persistent memory
C — Co-pilot	Yes	Yes (≥2 KB tool results stubbed in model's view)	Yes	Yes (in-session)	Auto-everything; agents that hit context limits

The Inspector primitives (classify, extract_metadata, scan) run locally with no cloud dependency. They're free under MIT.

from memosift.classifier import classify
from memosift.metadata import extract_metadata
from memosift.security import scan
from memosift.types import SecurityMode

ct = classify(my_tool_output)              # ContentType enum
md = extract_metadata(my_tool_output, ct)  # MetadataResult dataclass

# Security: FLAG (annotate), REDACT (mask matches), or BLOCK (raise)
result = scan(my_tool_output, mode=SecurityMode.REDACT)
print(result.content)                      # secrets masked
print(result.findings)                     # tuple of SecurityFinding

Cross-session memory, recall, compress, and explore (Modes B and C) require an API key.

---

Framework adapters

Each adapter leverages the most natural extension point of its target framework. Tool outputs flow through the MemoSift interception pipeline transparently — large results get replaced in place with artifact stubs (Mode C), or just observed and tracked (Modes A/B).

Anthropic — wrap_anthropic

Proxies messages.create / messages.stream and walks request content[] for tool_result blocks.

from anthropic import AsyncAnthropic
from memosift import MemoSift

ms = MemoSift(
    api_key=os.environ["MEMOSIFT_API_KEY"],
    base_url="https://dev.memosift.com",
)
client = ms.wrap_anthropic(AsyncAnthropic(), session_id="my-session")

# Use `client` exactly like the original. tool_result blocks larger than the
# threshold are silently swapped for artifact stubs before the model sees them.
response = await client.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=1024,
    messages=[...],
)

OpenAI — wrap_openai

Proxies both chat.completions.create (intercepts role="tool" messages) and responses.create (intercepts function_call_output items).

from openai import AsyncOpenAI

client = ms.wrap_openai(AsyncOpenAI(), session_id="my-session")

response = await client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {"role": "user", "content": "fetch data"},
        {"role": "tool", "name": "csv_fetch", "tool_call_id": "c1", "content": csv_data},
    ],
)

LangChain — langchain_callback

Returns an AsyncCallbackHandler you can register on any chain, agent, or tool. Uses LangChain's native callback protocol — no monkey-patching.

from langchain_openai import ChatOpenAI

handler = ms.langchain_callback(session_id="my-session")
llm = ChatOpenAI(callbacks=[handler])
# Or pass at runtime: llm.invoke(prompt, config={"callbacks": [handler]})

LangGraph — langgraph_wrapper

Returns an awrap_tool_call function for LangGraph's ToolNode.

from langgraph.prebuilt import ToolNode

tool_node = ToolNode(tools=my_tools, awrap_tool_call=ms.langgraph_wrapper())

Claude Agent SDK — claude_agent_hooks

Returns a PostToolUse hooks dict that injects MemoSift into the Claude Agent SDK lifecycle.

from claude_agent_sdk import ClaudeAgentOptions

options = ClaudeAgentOptions(hooks=ms.claude_agent_hooks())

OpenAI Agents SDK — openai_agents_tool_wrapper

Returns a decorator that wraps tool functions to intercept results.

intercept = ms.openai_agents_tool_wrapper(session_id="my-session")

@intercept
async def fetch_data(query: str) -> str:
    return long_csv_content

Generic adapter — wrap_generic

For frameworks that don't have a dedicated adapter. You declare which methods to intercept and supply an extract function.

from memosift.adapters import ExtractedToolResult

def extract(result, ctx):
    return [ExtractedToolResult(tool_name="my_tool", content=str(result))]

wrapped = ms.wrap_generic(
    my_client,
    session_id="my-session",
    methods=["chat.completions.create"],
    extract=extract,
)

---

Teach your AI coding agent how to use MemoSift

This SDK ships with an installable skill that teaches Claude Code, Cursor, Codex CLI, Copilot, or Windsurf how to wire MemoSift into your code correctly:

# After installing the SDK, run from your project root:
memosift install-skill

# Or install everywhere this project supports:
memosift install-skill --all

# Or install globally for Claude Code:
memosift install-skill --user

The CLI auto-detects which agents are configured (.claude/, .cursor/rules/, AGENTS.md) and writes the appropriate format for each. Idempotent — re-runs without --force skip files already containing the MemoSift section.

---

CLI

memosift login                 # save your API key to ~/.memosift/config.json
memosift install-skill         # install the agent skill into the current project
memosift install-claude-code   # wire MemoSift hooks into Claude Code settings
memosift serve                 # run as MCP server over stdio (for Claude Code)
memosift project create NAME   # create a new project on the cloud
memosift key create            # create a new API key
memosift sessions              # list sessions for a project
memosift usage                 # show usage and rate limits

---

Architecture

┌──────────────────────────────────────────────────┐
│              YOUR AGENT FRAMEWORK                │
│   (OpenAI / Anthropic / LangChain / Agent SDK …) │
└────────────────┬─────────────────────────────────┘
                 │ messages, tool results
                 ▼
┌──────────────────────────────────────────────────┐
│              MemoSift SDK (local)                │
│  Inspector primitives:                           │
│    classify → extract_metadata → scan            │
│  Sidecar / Co-pilot:                             │
│    track / recall / compress / explore / fetch   │
└────────────────┬─────────────────────────────────┘
                 │ HTTPS (only for Modes B/C)
                 ▼
┌──────────────────────────────────────────────────┐
│            MemoSift Cloud (paid)                 │
│  Phase 1 (≤3s inline):                           │
│    extract → embed → write_turn_records          │
│  Phase 2 (async):                                │
│    reconcile_memories / artifacts / entities     │
└──────────────────────────────────────────────────┘

Full design and the latest changelog are published at memosift.com/docs.

---

Verifying a deployment

import asyncio, os
from memosift import MemoSift

async def smoke():
    ms = MemoSift(
        api_key=os.environ["MEMOSIFT_API_KEY"],
        base_url="https://dev.memosift.com",
    )
    sid = "smoke-test"
    await ms.track([{"role": "user", "content": "hello memosift"}], session_id=sid)
    await asyncio.sleep(20)  # let Phase 2 reconciliation settle
    rec = await ms.recall(query="hello", session_id=sid, limit=5)
    print(f"memories: {rec.total_available}")

asyncio.run(smoke())

If this round-trips a non-zero memory count within ~25 seconds, you're wired in correctly.

A health probe is also available:

curl https://dev.memosift.com/v1/health

---

License

MIT — see LICENSE.

The MemoSift cloud service and dashboard are proprietary; the Inspector primitives in this SDK run locally and are free to use under MIT with no API key required.

---

Links

• Homepage: https://memosift.com
• Cloud API base URL: https://dev.memosift.com
• Support: support@memosift.com