PYTHON SDK
Python SDK Reference.
The memosift package on PyPI. Tool-result interception, persistent memory, and 7 framework adapters in one install.
pip install memosiftPick your mode
Compare in detailA — Inspector
Local-only primitives — classify, extract metadata, scan. No cloud, no API key.
Compliance · telemetry · routing
B — Sidecar
Track turns to the cloud, recall across sessions. No tool-result mutation.
Long-running agents · cross-session memory
C — Co-pilot
Track + recall + auto-stub large tool results in the model's view (≥2 KB).
Auto-everything · context-budget driven
By mode — code patterns
The same SDK serves all three integration modes. Pick the one your agent needs and use the matching pattern.
Inspector
Local-only primitives — no cloud, no API key. Free under MIT.
# Pure local — no MemoSift instance needed
from memosift.classifier import classify
from memosift.metadata import extract_metadata
from memosift.security import scan
from memosift.types import SecurityMode
content_type = classify(my_tool_output)
metadata = extract_metadata(my_tool_output, content_type)
result = scan(my_tool_output, mode=SecurityMode.REDACT)
print(result.content) # secrets masked
print(result.findings) # list of SecurityFindingSidecar
Push turns to the cloud, recall across sessions. No tool-result mutation.
# Track + recall against the MemoSift cloud
import os
from memosift import MemoSift
ms = MemoSift(
api_key=os.environ["MEMOSIFT_API_KEY"],
base_url="https://dev.memosift.com",
)
# After each turn:
await ms.track(messages, session_id="my-session")
# When you need recall:
result = await ms.recall(query="user's question", session_id="my-session")
for item in result.items:
print(item.content[:120])Co-pilot
Adapter wraps the LLM client; large tool results auto-stub in the model's view.
# Tool results above ~2 KB are silently replaced with artifact stubs
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 Anthropic SDK.
response = await client.messages.create(
model="claude-sonnet-4-5",
max_tokens=1024,
messages=[...],
)MemoSift()
#MemoSift(api_key: str, base_url: str = "https://dev.memosift.com", timeout: float = 30.0)Create a MemoSift client. The API key is required and can be generated from the dashboard or CLI. All subsequent calls use this client instance.
from memosift import MemoSift
ms = MemoSift(api_key="msk_live_abc123")
# Self-hosted
ms = MemoSift(
api_key="msk_live_abc123",
base_url="https://memosift.internal.corp",
timeout=60.0,
)Returns
MemoSift
- session() — Create session-bound client
- intercept() — Intercept tool results
- recall() — Hybrid memory recall
session()
#session(session_id: str) → MemoSiftSessionReturns a session-bound client that pre-fills session_id on all subsequent calls. Recommended for most workflows.
session = ms.session("conv_2024_04_18_a")
# All calls now use session_id automatically
result = await session.intercept(content, tool_name="bash")
await session.track(messages)
memories = await session.recall("auth decisions")Returns
MemoSiftSession
- session_id — The bound session identifier
intercept()
#intercept(content: str | bytes, *, tool_name: str, session_id: str,
security_mode: SecurityMode = REDACT,
security_config: SecurityConfig | None = None) → InterceptResultRuns the 5-stage local pipeline on tool output: strip reasoning, security scan, size gate, classify, and metadata extraction. Content under 500 chars passes through unchanged. Content 500+ chars creates an artifact. Content 2000+ chars is replaced with a summary.
result = await ms.intercept(
tool_output,
tool_name="read_file",
session_id="my-session",
security_mode=SecurityMode.REDACT,
)
# result.agent_facing_content → summary or raw
# result.replaced → True if content was replaced
# result.artifact → ArtifactRecord if stored
# result.content_type → ContentType enum
# result.security_findings → tuple of SecurityFindingReturns
InterceptResult
- agent_facing_content — The processed content to return to the agent
- replaced — True if content was replaced with a summary
- artifact — ArtifactRecord if content was stored
- content_type — Classified ContentType enum
- security_findings — Tuple of SecurityFinding objects
- metadata — Extracted metadata dict
track()
#track(messages: list[dict], *, session_id: str,
intercept_results: Sequence[InterceptResult] | None = None) → NoneShips a conversation turn to the MemoSift cloud. If the cloud is unreachable, the turn is queued locally (up to 1000 items) and retried automatically. Attach intercept results to link artifacts with the turn.
await ms.track(
[
{"role": "user", "content": "Read the auth module"},
{"role": "assistant", "content": "Here's the code..."},
],
session_id="my-session",
intercept_results=[result],
)Returns
None
recall()
#recall(query: str | None = None, *, session_id: str, limit: int = 10,
scope: str = "session", mode: str = "fast",
filters: RecallFilters | None = None,
intent_weights: dict[int, float] | None = None,
query_embedding: list[float] | None = None,
intent_version: int | None = None,
turn: int | None = None) → RecallResultRetrieves memories from the session or project. Supports three mutually exclusive modes: query (text search), intent_version (goal epoch retrieval), and turn (temporal snapshot). Exactly one mode parameter must be provided.
# Query mode
result = await ms.recall(
"auth middleware decisions",
session_id="s1",
scope="project",
mode="deep",
)# Intent mode
result = await ms.recall(intent_version=2, session_id="s1")# Temporal mode
result = await ms.recall(turn=15, session_id="s1")Returns
RecallResult
- items — List of RecallItem matches
- total_available — Total matching items (before limit)
- session_intent — Current session intent summary
- diagnostics — Timing and scoring diagnostics
explore()
#explore(item_id: str, *, kind: Literal["memory", "artifact", "proposition"],
session_id: str, limit: int = 10) → ExploreResultMulti-axis graph exploration from a given anchor item. Returns connected entities, related artifacts, temporal neighbors, and intent connections.
related = await ms.explore(
item_id="mem_4f8a2c",
kind="memory",
session_id="s1",
limit=5,
)
for item in related.items:
print(f"{item.via}: {item.content[:80]}")Returns
ExploreResult
- anchor_id — ID of the anchor item
- anchor_kind — Type of the anchor item
- items — Connected items, each with a via field indicating the connection axis
fetch()
#fetch(artifact_id: str) → FetchResultRetrieves artifact metadata and content. For artifacts under 500KB, the content is returned inline as base64. For larger artifacts, a signed download URL is provided.
content = await ms.fetch("art_7f2a3b")
if content.bytes_b64:
raw = base64.b64decode(content.bytes_b64)
elif content.download_url:
# Stream large artifacts
async with httpx.AsyncClient() as client:
resp = await client.get(content.download_url)Returns
FetchResult
- artifact_id — The artifact identifier
- artifact_type — MIME type or content classification
- byte_size — Size in bytes
- download_url — Signed URL for large artifacts (>500KB)
- bytes_b64 — Base64-encoded content for small artifacts
- summary — Generated summary of the artifact
compress()
#compress(*, session_id: str) → CompressResultAssembles a compressed context window for the session. Uses pure SQL and template rendering with no LLM calls. Completes in under 50ms.
compressed = await ms.compress(session_id="s1")
# Use as system prompt or prepend to messages
system_msg = {"role": "system", "content": compressed.rendered_text}
# Or use the structured representation
for section in compressed.structured:
print(f"{section.label}: {len(section.content)} chars")Returns
CompressResult
- rendered_text — Ready-to-use compressed context string
- structured — Structured sections with labels and content
- diagnostics — Assembly timing and stats
get_at_turn()
#get_at_turn(session_id: str, turn: int) → SessionSnapshotBi-temporal snapshot of the session at a specific turn. Returns the intent that was active, memories that existed, and artifacts that had been seen up to that point.
snapshot = await ms.get_at_turn("s1", turn=15)
print(snapshot.intent_at_turn)
print(f"Memories active: {len(snapshot.memories_active_at_turn)}")
print(f"Artifacts seen: {len(snapshot.artifacts_seen_by_turn)}")Returns
SessionSnapshot
- intent_at_turn — The intent version active at this turn
- memories_active_at_turn — Memories that existed at this turn
- artifacts_seen_by_turn — Artifacts created or referenced by this turn
get_intent()
#get_intent(*, session_id: str) → SessionIntentRetrieves the full intent history for a session, including all versions and the turn each was derived at. Useful for understanding how the conversation goal evolved.
intent = await ms.get_intent(session_id="s1")
for version in intent.versions:
print(f"v{version.version} (turn {version.derived_at_turn}): {version.summary}")Returns
SessionIntent
- versions — List of intent versions with summary and derived_at_turn
Framework adapters
Each adapter wires MemoSift into the framework's native extension point \u2014 proxies for OpenAI / Anthropic, callback handlers for LangChain, hook wrappers for Claude Agent SDK, and so on. Tool results flow through interception transparently.
Anthropic
ms.wrap_anthropic(...)pip install "memosift[anthropic]"
Proxy on messages.create / messages.stream. Walks tool_result content blocks pre-flight; mutates oversized blocks in place with artifact stubs.
from anthropic import AsyncAnthropic
from memosift import MemoSift
ms = MemoSift(api_key="msk_...", base_url="https://dev.memosift.com")
client = ms.wrap_anthropic(AsyncAnthropic(), session_id="my-session")
response = await client.messages.create(
model="claude-sonnet-4-5",
max_tokens=1024,
messages=[...],
)OpenAI
ms.wrap_openai(...)pip install "memosift[openai]"
Proxies both chat.completions.create (intercepts role=tool messages) AND responses.create (intercepts function_call_output entries).
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},
],
)OpenAI Agents SDK
ms.openai_agents_tool_wrapper(...)Decorator that wraps async tool functions to intercept their outputs.
intercept = ms.openai_agents_tool_wrapper(session_id="my-session")
@intercept
async def fetch_data(query: str) -> str:
return long_csv_contentClaude Agent SDK
ms.claude_agent_hooks(...)Returns a PostToolUse hooks dict \u2014 the Claude Agent SDK's native hook system.
from claude_agent_sdk import ClaudeAgentOptions
options = ClaudeAgentOptions(hooks=ms.claude_agent_hooks())LangChain
ms.langchain_callback(...)pip install "memosift[langchain]"
Returns an AsyncCallbackHandler \u2014 LangChain's native callback protocol. Pass via callbacks=[...] to any chain, agent, or tool.
from langchain_openai import ChatOpenAI
handler = ms.langchain_callback(session_id="my-session")
llm = ChatOpenAI(callbacks=[handler])LangGraph
ms.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())Generic
ms.wrap_generic(...)Configurable Proxy-based wrapper for any client. 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,
)TYPES
Key Types
Data classes returned by SDK methods. All fields are typed and accessible as attributes.