Ctx Claude Mem¶

ctx_claude_mem — mvp.ctx_claude_mem

Cluster: Uncategorised | Type: component | MCP Tools: 26

Overview¶

Adapter for an operator-managed upstream claude-mem worker. It stores and retrieves Claude project observations, summaries, prompt templates, and session metadata through the worker API, and exposes health checks so a user can tell whether memory is actually available.

This component is useful when the claude-mem service is installed and reachable. It does not bundle a separate memory engine, so launch workflows should run health before depending on cross-session recall.

No-worker or offline operation is explicitly degraded, not silently complete. Degraded paths expose completion_state, warning_card, evidence, degraded, degradation_reason, and available_with so callers can distinguish useful memory results from dependency failures.

Capability discovery is available through info / claude_mem_info in metadata.capabilities. It reports worker reachability, endpoint inventory, auth/config disclosure, search modes, endpoint health, local store counts, and no-worker setup instructions. Persistent-state operations are disclosed as auth_required=True and auth_enforced=False until adapter RBAC exists.

observation_search and search may include agentic_evidence for the advisory returned-set rerank path. The rerank can only reorder or drop records from the worker-returned set. list_patterns is a block-dispatch introspection op, not a FastMCP data-plane tool.

Vendored source is not automatically redistributable

G6 keeps a pinned upstream claude-mem source snapshot at vendor/claude-mem so testing and packaging work can continue if upstream becomes unavailable. The root package is AGPL-3.0, and upstream's ragtime/ subtree carries PolyForm Noncommercial 1.0.0. Treat the vendored copy as source availability and test infrastructure unless a commercial bundle explicitly excludes or reviews restricted subtrees and presents the required license notices.

For local validation, use python scripts\claude_mem_vendor.py check, worker start, contract-test, and worker stop. Do not run the upstream global installer silently from G6 because it can rewrite user IDE hooks.

When to use:

Persisting Claude Code session observations between conversations
Storing and retrieving reusable prompt templates for Claude workflows
Building a searchable long-term memory for Claude-based pipelines
Checking whether the local claude-mem worker is reachable before a workflow starts

Example:

from mvp.ctx_claude_mem import CtxClaudeMemBlock, MemInput

block = CtxClaudeMemBlock(name="claude_mem")
health = block.infer(MemInput(op="health"))
result = block.infer(MemInput(op="observe", text="User prefers concise answers."))

Works well with: ctx_rag, align_prompt_library, ctx_recursive

Public API¶

`MemoryRerankDecision`¶

Validated advisory rerank verdict over a returned memory set.

Field	Type	Default
`ordered_indices`	`tuple[int, ...]`	required
`dropped_indices`	`tuple[int, ...]`	()
`rationale`	`str`	''
`eligible_fingerprint`	`str`	''
`confidence`	`float`	0.0
`degraded`	`bool`	False
`raw_response`	`str`	''

`LLMMemoryRerankRuntime`¶

Provider-neutral memory-rerank runtime backed by G6's LLM caller.

Constructor:

Parameter	Type	Default
`llm`	`LLMCaller \\| None`	None

Methods:

`rerank(query: str, memories: list[Any]) -> MemoryRerankDecision`¶

`CtxClaudeMemRerankPatternRuntime`¶

Stateless, load-bearing returned-set-ceiling enforcement.

Methods:

`enforce_eligibility(decision: MemoryRerankDecision, memories: list[Any]) -> tuple[list[Any], bool, bool]`¶

`CtxClaudeMemPlanner`¶

Runtime-first advisory memory-rerank facade with returned-order fallback.

Constructor:

Parameter	Type	Default
`runtime`	`MemoryRerankRuntime \\| None`	None
`pattern_runtime`	`CtxClaudeMemRerankPatternRuntime \\| None`	None

Methods:

`rerank(query: str, memories: list[Any]) -> list[Any]`¶

`CtxClaudeMemBlock(AIBlock[MemInput, MemOutput, None])`¶

5-op claude-mem block for pipeline composition.

Field	Type	Default
`name`	`str`	'ctx_claude_mem'
`state`	`None`	None
`base_url`	`str`	field(default_factory=default_base_url)
`timeout`	`float`	field(default_factory=lambda: float(os.environ.get('CLAUDE_MEM_TIMEOUT_SEC', '10')))
`resource_bounds`	`ResourceBounds \\| None`	None
`usage`	`ResourceUsage`	field(default_factory=ResourceUsage)
`agentic_planner`	`CtxClaudeMemPlanner \\| None`	None

Methods:

`infer(data: MemInput) -> Result[MemOutput]`¶

`MemInput(BaseModel)`¶

Input to CtxClaudeMemBlock — 5 high-level ops.

Field	Type	Default
`op`	`Literal['observe', 'recall', 'search', 'session_info', 'health']`	required
`text`	`str`	''
`query`	`str`	''
`observation_type`	`str`	''
`project`	`str`	''
`session_id`	`str`	''
`tags`	`list[str]`	Field(default_factory=list)
`limit`	`int`	10
`agentic_rerank`	`bool \\| None`	None
`run_mode`	`Literal['beta', 'production']`	'beta'
`reviewer_signature`	`str`	''

`MemOutput(BaseModel)`¶

Output from CtxClaudeMemBlock.

Field	Type	Default
`op`	`str`	required
`found`	`bool`	False
`count`	`int`	0
`records`	`list[dict]`	Field(default_factory=list)
`message`	`str`	''
`metadata`	`dict`	Field(default_factory=dict)
`degraded`	`bool`	False
`degradation_reason`	`str`	''
`available_with`	`str`	''
`completion_state`	`str`	'complete'
`warning_card`	`dict`	Field(default_factory=dict)
`evidence`	`dict`	Field(default_factory=dict)
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''
`agentic_evidence`	`dict`	Field(default_factory=dict)

`CtxClaudeMemMCPBlock(AIBlock[MCPClaudeMemInput, MCPClaudeMemOutput, dict])`¶

25-op claude-mem MCP block with SQLite persistence.

Field	Type	Default
`name`	`str`	'ctx_claude_mem_mcp'
`state`	`dict`	field(default_factory=dict)
`db_path`	`str`	field(default_factory=lambda: os.environ.get('CLAUDE_MEM_STORE_PATH', _DEFAULT_DB))
`base_url`	`str`	field(default_factory=default_base_url)
`timeout`	`float`	field(default_factory=lambda: float(os.environ.get('CLAUDE_MEM_TIMEOUT_SEC', '10')))
`max_retries`	`int`	field(default_factory=lambda: int(os.environ.get('CLAUDE_MEM_MAX_RETRIES', '3')))
`resource_bounds`	`ResourceBounds`	field(default_factory=ResourceBounds)
`usage`	`ResourceUsage`	field(default_factory=ResourceUsage)
`agentic_planner`	`CtxClaudeMemPlanner \\| None`	None

Methods:

`infer(data: MCPClaudeMemInput) -> Result[MCPClaudeMemOutput]`¶

`MCPClaudeMemRecord(BaseModel)`¶

A single record returned from claude-mem queries.

Field	Type	Default
`id`	`str`	''
`record_type`	`str`	''
`content`	`str`	''
`tags`	`list[str]`	Field(default_factory=list)
`timestamp`	`str`	''
`metadata`	`dict`	Field(default_factory=dict)

`MCPClaudeMemInput(BaseModel)`¶

Input to CtxClaudeMemMCPBlock — 25 ops.

Field	Type	Default
`op`	`Literal['session_create', 'session_get', 'session_list', 'session_files', 'observation_store', 'observation_get', 'observation_list', 'observation_search', 'observation_batch', 'summary_store', 'summary_get', 'summary_list', 'summary_search', 'prompt_save', 'prompt_get', 'prompt_list', 'prompt_search', 'timeline_around_timestamp', 'timeline_around_observation', 'search', 'projects_list', 'health', 'worker_start', 'worker_stop', 'info', 'list_patterns']`	required
`session_id`	`str`	''
`project`	`str`	''
`description`	`str`	''
`status`	`str`	''
`observation_id`	`str`	''
`content`	`str`	''
`observation_type`	`str`	''
`tags`	`list[str]`	Field(default_factory=list)
`observations`	`list[dict]`	Field(default_factory=list)
`summary_text`	`str`	''
`prompt_number`	`int`	0
`prompt_text`	`str`	''
`role`	`str`	''
`anchor`	`str`	''
`anchor_observation`	`str`	''
`depth_before`	`int`	5
`depth_after`	`int`	5
`query`	`str`	''
`limit`	`int`	50
`worker_path`	`str`	''
`operator_consent`	`bool`	False
`run_mode`	`Literal['beta', 'production']`	'beta'
`reviewer_signature`	`str`	''

`MCPClaudeMemOutput(BaseModel)`¶

Output from CtxClaudeMemMCPBlock.

Field	Type	Default
`op`	`str`	required
`found`	`bool`	False
`count`	`int`	0
`records`	`list[dict]`	Field(default_factory=list)
`retrieved`	`list[dict]`	Field(default_factory=list)
`message`	`str`	''
`metadata`	`dict`	Field(default_factory=dict)
`degraded`	`bool`	False
`degradation_reason`	`str`	''
`available_with`	`str`	''
`completion_state`	`str`	'complete'
`warning_card`	`dict`	Field(default_factory=dict)
`evidence`	`dict`	Field(default_factory=dict)
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''
`agentic_evidence`	`dict`	Field(default_factory=dict)

`ClaudeMemStore`¶

SQLite-backed local store for the ctx_claude_mem MCP sub-package.

Constructor:

Parameter	Type	Default
`db_path`	`str`	':memory:'

Methods:

`log_request(op: str, path: str = '', status_code: int = 0, latency_ms: float = 0.0, error: str = '') -> None`¶

`list_requests(limit: int = 50) -> list[dict]`¶

`cache_get(cache_key: str) -> dict | None`¶

`cache_set(cache_key: str, response: dict, ttl_seconds: int = 300) -> None`¶

`update_worker_status(is_alive: bool, pid: int = 0, port: int = 0) -> None`¶

`get_worker_status() -> dict | None`¶

`map_session(g6_session_id: str, cm_session_id: str) -> None`¶

`get_cm_session(g6_session_id: str) -> str | None`¶

`log_error(op: str, error_message: str, error_type: str = '', request_json: dict | None = None) -> None`¶

`list_errors(limit: int = 50) -> list[dict]`¶

`count_all() -> dict[str, int]`¶

`text_search(query: str, top_k: int = 10) -> list[dict]`¶

Search request_log + errors by TF-IDF (falls back to substring).

Functions¶

`agentic_planner_enabled(default_enabled: bool) -> bool`¶

Decide whether the agentic memory-rerank planner should be used.

`validate_memory_rerank_decision(decision: MemoryRerankDecision, n_eligible: int, expected_fingerprint: str) -> None`¶

Returned-set / anti-injection guard for a memory-rerank decision.

`applied_agentic_patterns() -> list[dict[str, Any]]`¶

Return compact metadata for ctx_claude_mem-applied vendored patterns.

`get_skill_catalog() -> CtxClaudeMemSkillCatalog`¶

MCP Tools¶

Operation	Source
`session_create`	claude_mem_mcp
`session_get`	claude_mem_mcp
`session_list`	claude_mem_mcp
`session_files`	claude_mem_mcp
`observation_store`	claude_mem_mcp
`observation_get`	claude_mem_mcp
`observation_list`	claude_mem_mcp
`observation_search`	claude_mem_mcp
`observation_batch`	claude_mem_mcp
`summary_store`	claude_mem_mcp
`summary_get`	claude_mem_mcp
`summary_list`	claude_mem_mcp
`summary_search`	claude_mem_mcp
`prompt_save`	claude_mem_mcp
`prompt_get`	claude_mem_mcp
`prompt_list`	claude_mem_mcp
`prompt_search`	claude_mem_mcp
`timeline_around_timestamp`	claude_mem_mcp
`timeline_around_observation`	claude_mem_mcp
`search`	claude_mem_mcp
`projects_list`	claude_mem_mcp
`health`	claude_mem_mcp
`worker_start`	claude_mem_mcp
`worker_stop`	claude_mem_mcp
`info`	claude_mem_mcp
`list_patterns`	claude_mem_mcp

Ctx Claude Mem¶

Overview¶

Public API¶

MemoryRerankDecision¶

LLMMemoryRerankRuntime¶

rerank(query: str, memories: list[Any]) -> MemoryRerankDecision¶

CtxClaudeMemRerankPatternRuntime¶

enforce_eligibility(decision: MemoryRerankDecision, memories: list[Any]) -> tuple[list[Any], bool, bool]¶

CtxClaudeMemPlanner¶

rerank(query: str, memories: list[Any]) -> list[Any]¶

CtxClaudeMemBlock(AIBlock[MemInput, MemOutput, None])¶

infer(data: MemInput) -> Result[MemOutput]¶

MemInput(BaseModel)¶

MemOutput(BaseModel)¶

CtxClaudeMemMCPBlock(AIBlock[MCPClaudeMemInput, MCPClaudeMemOutput, dict])¶

infer(data: MCPClaudeMemInput) -> Result[MCPClaudeMemOutput]¶

MCPClaudeMemRecord(BaseModel)¶

MCPClaudeMemInput(BaseModel)¶

MCPClaudeMemOutput(BaseModel)¶

ClaudeMemStore¶

log_request(op: str, path: str = '', status_code: int = 0, latency_ms: float = 0.0, error: str = '') -> None¶

list_requests(limit: int = 50) -> list[dict]¶

cache_get(cache_key: str) -> dict | None¶

cache_set(cache_key: str, response: dict, ttl_seconds: int = 300) -> None¶

update_worker_status(is_alive: bool, pid: int = 0, port: int = 0) -> None¶

get_worker_status() -> dict | None¶

map_session(g6_session_id: str, cm_session_id: str) -> None¶

get_cm_session(g6_session_id: str) -> str | None¶

log_error(op: str, error_message: str, error_type: str = '', request_json: dict | None = None) -> None¶

list_errors(limit: int = 50) -> list[dict]¶

count_all() -> dict[str, int]¶

text_search(query: str, top_k: int = 10) -> list[dict]¶

Functions¶

agentic_planner_enabled(default_enabled: bool) -> bool¶

validate_memory_rerank_decision(decision: MemoryRerankDecision, n_eligible: int, expected_fingerprint: str) -> None¶

applied_agentic_patterns() -> list[dict[str, Any]]¶

get_skill_catalog() -> CtxClaudeMemSkillCatalog¶

MCP Tools¶

`MemoryRerankDecision`¶

`LLMMemoryRerankRuntime`¶

`rerank(query: str, memories: list[Any]) -> MemoryRerankDecision`¶

`CtxClaudeMemRerankPatternRuntime`¶

`enforce_eligibility(decision: MemoryRerankDecision, memories: list[Any]) -> tuple[list[Any], bool, bool]`¶

`CtxClaudeMemPlanner`¶

`rerank(query: str, memories: list[Any]) -> list[Any]`¶

`CtxClaudeMemBlock(AIBlock[MemInput, MemOutput, None])`¶

`infer(data: MemInput) -> Result[MemOutput]`¶

`MemInput(BaseModel)`¶

`MemOutput(BaseModel)`¶

`CtxClaudeMemMCPBlock(AIBlock[MCPClaudeMemInput, MCPClaudeMemOutput, dict])`¶

`infer(data: MCPClaudeMemInput) -> Result[MCPClaudeMemOutput]`¶

`MCPClaudeMemRecord(BaseModel)`¶

`MCPClaudeMemInput(BaseModel)`¶

`MCPClaudeMemOutput(BaseModel)`¶

`ClaudeMemStore`¶

`log_request(op: str, path: str = '', status_code: int = 0, latency_ms: float = 0.0, error: str = '') -> None`¶

`list_requests(limit: int = 50) -> list[dict]`¶

`cache_get(cache_key: str) -> dict | None`¶

`cache_set(cache_key: str, response: dict, ttl_seconds: int = 300) -> None`¶

`update_worker_status(is_alive: bool, pid: int = 0, port: int = 0) -> None`¶

`get_worker_status() -> dict | None`¶

`map_session(g6_session_id: str, cm_session_id: str) -> None`¶

`get_cm_session(g6_session_id: str) -> str | None`¶

`log_error(op: str, error_message: str, error_type: str = '', request_json: dict | None = None) -> None`¶

`list_errors(limit: int = 50) -> list[dict]`¶

`count_all() -> dict[str, int]`¶

`text_search(query: str, top_k: int = 10) -> list[dict]`¶

`agentic_planner_enabled(default_enabled: bool) -> bool`¶

`validate_memory_rerank_decision(decision: MemoryRerankDecision, n_eligible: int, expected_fingerprint: str) -> None`¶

`applied_agentic_patterns() -> list[dict[str, Any]]`¶

`get_skill_catalog() -> CtxClaudeMemSkillCatalog`¶