PYTHON SDK
SDK Reference
Complete reference for the MemoSift Python SDK. Intercept tool results, track conversations, recall memories, and explore the knowledge graph.
pip install memosiftMemoSift()
#MemoSift(api_key: str, base_url: str = "https://api.memosift.dev", 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
ADAPTERS
Framework Integrations
Drop-in wrappers for popular agent frameworks. Each adapter handles tool interception, turn tracking, and memory injection automatically.
openai_agents_tool_wrapper()
#openai_agents_tool_wrapper(*, session_id: str) → CallableReturns a decorator that wraps OpenAI Agents SDK tool functions. Automatically intercepts tool output and tracks turns.
wrap = ms.openai_agents_tool_wrapper(session_id="my-session")
@wrap
async def read_file(path: str) -> str:
with open(path) as f:
return f.read()Returns
Callable
- (decorator) — Wrap any async tool function to add interception
langgraph_wrapper()
#langgraph_wrapper() → CallableReturns a wrapper compatible with LangGraph's awrap_tool_call parameter. The session ID is automatically extracted from config.configurable.thread_id.
No parameters required
from langgraph.prebuilt import ToolNode
awrap = ms.langgraph_wrapper()
tool_node = ToolNode(tools=[...], awrap_tool_call=awrap)
# session_id auto-extracted from config.configurable.thread_idReturns
Callable
- (wrapper) — Compatible with ToolNode's awrap_tool_call parameter
claude_agent_hooks()
#claude_agent_hooks() → dictReturns a hooks dictionary for Anthropic's Agent SDK. Large tool results are automatically stored as artifacts and replaced with a fetch reference.
No parameters required
hooks = ms.claude_agent_hooks()
# Returns {"PostToolUse": async_handler}
# Large results are replaced with:
# "[MemoSift] Stored as artifact art_abc. Use memosift_fetch('art_abc')"Returns
dict
- PostToolUse — Async handler for post-tool-use interception
wrap()
#wrap(client, *, session_id: str) → WrappedClientWraps an OpenAI client for full lifecycle interception. Before each call, memories are recalled and injected. During tool use, large results are intercepted. After each call, the turn is tracked in the background. Context is auto-compressed at 80% window pressure.
from openai import AsyncOpenAI
wrapped = ms.wrap(AsyncOpenAI(), session_id="my-session")
# Pre-call: recall + inject memories
# Tool use: intercept large results
# Post-call: track in background
# Auto-compress at 80% context pressureReturns
WrappedClient
- (client) — Drop-in replacement with automatic memory management
tools()
#tools(*, session_id: str) → list[dict]Returns 4 OpenAI-compatible function schemas that agents can call directly: memosift_recall, memosift_explore, memosift_fetch, and memosift_compress.
tool_schemas = ms.tools(session_id="my-session")
# Pass to OpenAI
response = await client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tool_schemas,
)Returns
list[dict]
- (schemas) — 4 OpenAI function-calling tool definitions
TYPES
Key Types
Data classes returned by SDK methods. All fields are typed and accessible as attributes.