Ctx Claude Context¶

Ctx Claude Context — mvp.ctx_claude_context

Cluster: Context & Retrieval | Type: component | MCP Tools: 26

Overview¶

Sliding context window manager for Claude conversations, tracking token counts via tiktoken (chars/4 heuristic fallback) and providing compress and trim operations to keep the conversation within a token budget. Compression replaces the oldest messages with an extractive summary; trim performs a single O(n) slice to drop oldest messages until the total fits.

When to use:

• Managing long Claude conversation histories without exceeding the context window
• Automatically summarising older turns before appending new messages
• Maintaining a system prompt alongside a rolling message window

Example:

from mvp.ctx_claude_context import CtxClaudeContextBlock, ClaudeContextInput

block = CtxClaudeContextBlock(name="ctx_cc")
block.infer(ClaudeContextInput(operation="add", role="user", content="Hello, world!"))
result = block.infer(ClaudeContextInput(operation="trim", max_tokens=2000))
# result.value.total_tokens → token count after trim

Works well with: ctx_ace, ctx_recursive, llm_router

Public API¶

CtxClaudeContextDecisionError(ValueError)¶

The LLM did not produce a usable, validated retention decision.

ContextRetentionDecision¶

Validated advisory retention verdict over an existing message set.

LLMContextRetentionRuntime¶

Provider-neutral retention runtime backed by G6's LLM caller interface.

Constructor:

Methods:

select(messages: list[tuple[str, str]], token_counts: list[int], budget: int) -> ContextRetentionDecision¶

CtxClaudeContextRetentionPatternRuntime¶

Stateless, load-bearing budget-ceiling enforcement.

Methods:

enforce_budget(decision: ContextRetentionDecision, messages: list[tuple[str, str]], token_counts: list[int], budget: int, recency_retained: list[int]) -> tuple[list[int], list[tuple[str, str]], bool, bool]¶

CtxClaudeContextPlanner¶

Runtime-first advisory retention facade with deterministic fallback.

Constructor:

Methods:

select(messages: list[tuple[str, str]], token_counts: list[int], budget: int) -> list[int]¶

Return the retained message indices (chronological) under the budget.

CtxClaudeContextBlock(AIBlock[ClaudeContextInput, ClaudeContextOutput, dict])¶

Sliding context window manager for Claude conversations.

Methods:

infer(data: ClaudeContextInput) -> Result[ClaudeContextOutput]¶

ContextMessage(BaseModel)¶

ClaudeContextInput(BaseModel)¶

Methods:

max_tokens_positive(v: int) -> int¶

compress_ratio_valid(v: float) -> float¶

ClaudeContextOutput(BaseModel)¶

CtxClaudeContextMCPBlock(AIBlock[MCPContextInput, MCPContextOutput, dict])¶

25-op MCP block for context engineering with SQLite persistence.

Methods:

infer(data: MCPContextInput) -> Result[MCPContextOutput]¶

MCPContextInput(BaseModel)¶

Methods:

max_tokens_positive(v: int) -> int¶

compress_ratio_valid(v: float) -> float¶

MCPContextRecord(BaseModel)¶

MCPContextOutput(BaseModel)¶

ContextStore¶

SQLite store for context engineering persistence.

Constructor:

Methods:

save_session(name: str, messages_json: str, system_prompt: str, total_tokens: int, message_count: int, tags: list[str] | None = None, notes: str = '') -> int¶

load_session(name: str) -> dict | None¶

list_sessions(limit: int = 50, query: str = '') -> list[dict]¶

upsert_budget(name: str, total_limit: int, reserved_buffer: int = 0, category_limits_json: str = '{}') -> int¶

get_budget(name: str) -> dict | None¶

add_degradation_event(health_score: float, status: str, utilization: float, degradation_score: float, poisoning_risk: float, total_tokens: int, message_count: int, recommendations_json: str = '[]') -> int¶

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

clear_degradation_events() -> int¶

add_compression_record(method: str, messages_before: int, messages_after: int, tokens_before: int, tokens_after: int, quality_score: float = 0.0, probe_results_json: str = '{}') -> int¶

upsert_category(category: str, message_count: int, token_count: int, summary: str = '') -> int¶

get_categories() -> list[dict]¶

clear_categories() -> int¶

add_context_snapshot(session_name: str, messages_json: str, total_tokens: int, operation: str, notes: str = '') -> int¶

add_metric(metric_type: str, metric_value: float, details_json: str = '{}') -> int¶

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

Substring search across sessions, budgets, categories, contexts.

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

set_budget(name: str, total_limit: int, reserved_buffer: int = 0, category_limits_json: str = '{}') -> int¶

categorize() -> list[dict]¶

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

capabilities() -> dict¶

health() -> dict¶

translate_error(exc: Exception) -> dict¶

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

close() -> None¶

Functions¶

agentic_planner_enabled(default_enabled: bool) -> bool¶

Decide whether the agentic retention planner should be used.

messages_fingerprint(messages: list[tuple[str, str]]) -> str¶

sha256 over the existing (role, content) message set (order-sensitive).

validate_retention_decision(decision: ContextRetentionDecision, n_messages: int, expected_fingerprint: str) -> None¶

Structural / anti-injection guard for a retention decision (FAIL-CLOSED).

recency_floor_indices(token_counts: list[int], budget: int) -> list[int]¶

The deterministic RECENCY floor: the LARGEST FITTING SUFFIX under the budget.

planner_is_llm_trusted(planner: Any) -> bool¶

Whether the BLOCK may report llm_used=True for planner.

MCP Tools¶