Ctx Elastic¶

Ctx Elastic — mvp.ctx_elastic

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

Overview¶

Thin wrapper around the Elasticsearch Python client that performs match queries with optional term-filter pushdown and returns ranked SearchHit results. The client is built lazily on the first infer() call and reused, supporting API key auth, HTTP basic auth, and custom CA certificates for TLS. Returns a structured SearchResults object with hits, total count, and the index name used.

Production caveat: ctx_elastic is an optional Elasticsearch adapter, not an out-of-the-box universal search layer. It requires a reachable Elasticsearch cluster and an index whose mapping matches the query shape. The default search path matches against a content field; use index_doc, explicit mappings, natural_language_search, or a custom ES query path when your documents use different searchable fields. If Elasticsearch is absent, the component fails quickly with a diagnostic Result.fail() rather than falling back to local search.

Safety and degradation contract: MCP destructive/admin mutations (delete_doc, delete_index, reindex, update_by_query, and alias_manage with alias_action add or remove) require confirm=True and a non-empty reason; otherwise the operation returns completion_state: blocked with warning code G6_E_ELASTIC_DESTRUCTIVE_CONFIRMATION_REQUIRED before any Elasticsearch call is attempted. Non-clean MCP outcomes surface completion_state, warning_card, evidence, and the correlation fields request_id, task_id, and run_id while preserving degraded, degradation_reason, and agentic_evidence. Stable degradation codes cover missing package, unreachable cluster, auth/TLS, index missing, mapping mismatch, timeout, malformed JSON, empty result, and unimplemented safety gates.

Capability discovery: elastic_info() includes per-op metadata (risk_level, mutates_state, destructive, requires_elasticsearch, supports_dry_run, supports_confirmation, and expected_degradation_modes) so callers can plan safely before invoking tools. credential_authorization_verified: false is intentional: the production reviewer_signature is an acknowledgement artifact, not credential or promotion-authority verification.

When to use:

Searching large document collections stored in Elasticsearch with field-level filters
Integrating full-text retrieval into a RAG or grounding pipeline backed by an ES cluster
Replacing a mock search engine with a production-grade index in the same pipeline

Do not use as the first retrieval choice when:

You do not already have Elasticsearch running
You only need small local document search (ctx_rag is simpler)
Your index does not expose the fields expected by the selected search operation

Example:

from mvp.ctx_elastic import CtxElasticBlock, SearchQuery

block = CtxElasticBlock(name="elastic", host="my-es-host", default_index="docs")
result = block.infer(SearchQuery(query="neural networks", top_k=5))
# result.value.hits → list of SearchHit with id, score, source

Works well with: ctx_rag, ctx_colbert, ctx_mnm

Public API¶

`HitRerankDecision`¶

Validated advisory rerank verdict over a returned hit 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`	''

`LLMHitRerankRuntime`¶

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

Constructor:

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

Methods:

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

`CtxElasticRerankPatternRuntime`¶

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

Methods:

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

`CtxElasticPlanner`¶

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

Constructor:

Parameter	Type	Default
`runtime`	`HitRerankRuntime \\| None`	None
`pattern_runtime`	`CtxElasticRerankPatternRuntime \\| None`	None

Methods:

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

`CtxElasticBlock(AIBlock[SearchQuery, SearchResults, None])`¶

Searches an Elasticsearch index and returns ranked document hits.

Field	Type	Default
`name`	`str`	'ctx_elastic'
`host`	`str`	'localhost'
`port`	`int`	9200
`scheme`	`str`	'http'
`default_index`	`str`	'default'
`api_key`	`str`	''
`username`	`str`	''
`password`	`str`	''
`ca_certs`	`str`	''
`request_timeout`	`float`	0.25
`resource_bounds`	`ResourceBounds \\| None`	None
`usage`	`ResourceUsage`	field(default_factory=ResourceUsage)
`agentic_planner`	`CtxElasticPlanner \\| None`	None

Methods:

`infer(data: SearchQuery) -> Result[SearchResults]`¶

`SearchQuery(BaseModel)`¶

Input for CtxElasticBlock — a text search query.

Field	Type	Default
`query`	`str`	required
`index`	`str`	'default'
`top_k`	`int`	Field(default=10, le=100)
`filters`	`dict[str, Any]`	Field(default_factory=dict)
`agentic_rerank`	`bool \\| None`	None
`run_mode`	`Literal['beta', 'production']`	'beta'
`reviewer_signature`	`str`	''
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''

`SearchHit(BaseModel)`¶

A single document hit from an Elasticsearch query.

Field	Type	Default
`id`	`str`	required
`score`	`float`	required
`source`	`dict[str, Any]`	Field(default_factory=dict)

`SearchResults(BaseModel)`¶

Results from CtxElasticBlock.

Field	Type	Default
`hits`	`list[SearchHit]`	required
`total`	`int`	required
`index`	`str`	required
`query`	`str`	required
`degraded`	`bool`	False
`degradation_reason`	`str`	''
`completion_state`	`str`	'qualified-draft'
`warning_card`	`dict[str, Any]`	Field(default_factory=dict)
`evidence`	`dict[str, Any]`	Field(default_factory=dict)
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''
`agentic_evidence`	`dict[str, Any]`	Field(default_factory=dict)

`CtxElasticMCPBlock(AIBlock[MCPElasticInput, MCPElasticOutput, dict])`¶

Full-featured Elasticsearch MCP block with SQLite persistence.

Field	Type	Default
`name`	`str`	'ctx_elastic_mcp'
`state`	`dict \\| None`	None
`db_path`	`str`	':memory:'
`host`	`str`	'localhost'
`port`	`int`	9200
`scheme`	`str`	'http'
`api_key`	`str`	''
`username`	`str`	''
`password`	`str`	''
`ca_certs`	`str`	''
`request_timeout`	`float`	0.25
`resource_bounds`	`ResourceBounds \\| None`	None
`usage`	`ResourceUsage`	field(default_factory=ResourceUsage)
`agentic_planner`	`CtxElasticPlanner \\| None`	None

Methods:

`infer(data: MCPElasticInput) -> Result[MCPElasticOutput]`¶

`MCPElasticInput(BaseModel)`¶

Input for all 25 Elasticsearch MCP operations.

Field	Type	Default
`op`	`Literal['search', 'get_doc', 'index_doc', 'delete_doc', 'bulk_index', 'create_index', 'delete_index', 'aggregate', 'count', 'cluster_info', 'save_session', 'load_sessions', 'save_template', 'load_templates', 'save_bookmark', 'analyze_text', 'explain_query', 'reindex', 'update_by_query', 'field_caps', 'index_stats', 'alias_manage', 'natural_language_search', 'auto_mapping', 'search_explain', 'list_patterns']`	required
`index`	`str`	'default'
`query`	`str`	''
`doc_id`	`str`	''
`top_k`	`int`	Field(default=10, le=100)
`count`	`int`	Field(default=0, ge=0)
`filters_json`	`str`	'{}'
`doc_json`	`str`	'{}'
`docs_json`	`str`	'[]'
`mappings_json`	`str`	'{}'
`agg_type`	`str`	''
`field`	`str`	''
`name`	`str`	''
`description`	`str`	''
`tags`	`list[str]`	Field(default_factory=list)
`notes`	`str`	''
`results_json`	`str`	'[]'
`body_json`	`str`	'{}'
`index_pattern`	`str`	''
`session_id`	`str`	''
`analyzer`	`str`	'standard'
`text`	`str`	''
`script_json`	`str`	'{}'
`source_index`	`str`	''
`dest_index`	`str`	''
`indices`	`str`	'*'
`alias_name`	`str`	''
`alias_action`	`str`	''
`sample_docs_json`	`str`	'[]'
`doc_source_json`	`str`	'{}'
`limit`	`int`	Field(default=50, ge=1, le=500)
`relevance_score`	`float`	0.0
`confirm`	`bool`	False
`reason`	`str`	''
`clicked_doc_ids`	`list[str]`	Field(default_factory=list)
`agentic_rerank`	`bool \\| None`	None
`run_mode`	`Literal['beta', 'production']`	'beta'
`reviewer_signature`	`str`	''
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''

`MCPElasticRecord(BaseModel)`¶

A generic record returned from the SQLite store.

Field	Type	Default
`id`	`str`	''
`record_type`	`str`	''
`key`	`str`	''
`value`	`str`	''
`tags`	`list[str]`	Field(default_factory=list)
`timestamp`	`str`	''
`metadata`	`dict[str, Any]`	Field(default_factory=dict)

`MCPElasticOutput(BaseModel)`¶

Output for all 25 Elasticsearch MCP operations.

Field	Type	Default
`op`	`str`	''
`key`	`str`	''
`value`	`str`	''
`found`	`bool`	False
`count`	`int`	0
`records`	`list[MCPElasticRecord]`	Field(default_factory=list)
`results`	`list[dict[str, Any]]`	Field(default_factory=list)
`summary`	`str`	''
`message`	`str`	''
`data_json`	`str`	'{}'
`metadata`	`dict[str, Any]`	Field(default_factory=dict)
`degraded`	`bool`	False
`degradation_reason`	`str`	''
`completion_state`	`str`	'qualified-draft'
`warning_card`	`dict[str, Any]`	Field(default_factory=dict)
`evidence`	`dict[str, Any]`	Field(default_factory=dict)
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''
`agentic_evidence`	`dict[str, Any]`	Field(default_factory=dict)

`ElasticStore`¶

Sync SQLite store with 6 tables.

Constructor:

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

Methods:

`save_session(index_name: str = 'default', query_text: str = '', query_json: str = '{}', results_json: str = '[]', hit_count: int = 0, tags: list[str] | None = None, notes: str = '') -> str`¶

`get_sessions(index_name: str = '', tag: str = '', limit: int = 50) -> list[dict[str, Any]]`¶

`save_template(name: str, description: str = '', body_json: str = '{}', index_pattern: str = '', tags: list[str] | None = None) -> str`¶

`get_template(name: str) -> dict[str, Any] | None`¶

`get_templates(tag: str = '', search: str = '', limit: int = 50) -> list[dict[str, Any]]`¶

`save_bookmark(index_name: str = 'default', doc_id: str = '', doc_source_json: str = '{}', notes: str = '', tags: list[str] | None = None) -> str`¶

`get_bookmarks(index_name: str = '', tag: str = '', limit: int = 50) -> list[dict[str, Any]]`¶

`save_outcome(session_id: str = '', query_text: str = '', relevance_score: float = 0.0, clicked_doc_ids: list[str] | None = None, notes: str = '') -> str`¶

`cache_llm(prompt_text: str, response_text: str, model: str = '') -> str`¶

`get_cached_llm(prompt_text: str) -> str | None`¶

`get_config(key: str) -> str | None`¶

`set_config(key: str, value_json: str) -> None`¶

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

Functions¶

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

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

`validate_hit_rerank_decision(decision: HitRerankDecision, n_eligible: int, expected_fingerprint: str) -> None`¶

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

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

Return compact metadata for ctx_elastic-applied vendored patterns.

`get_skill_catalog() -> CtxElasticSkillCatalog`¶

MCP Tools¶

Operation	Source
`search`	elastic_mcp
`get_doc`	elastic_mcp
`index_doc`	elastic_mcp
`delete_doc`	elastic_mcp
`bulk_index`	elastic_mcp
`create_index`	elastic_mcp
`delete_index`	elastic_mcp
`aggregate`	elastic_mcp
`count`	elastic_mcp
`cluster_info`	elastic_mcp
`save_session`	elastic_mcp
`load_sessions`	elastic_mcp
`save_template`	elastic_mcp
`load_templates`	elastic_mcp
`save_bookmark`	elastic_mcp
`analyze_text`	elastic_mcp
`explain_query`	elastic_mcp
`reindex`	elastic_mcp
`update_by_query`	elastic_mcp
`field_caps`	elastic_mcp
`index_stats`	elastic_mcp
`alias_manage`	elastic_mcp
`natural_language_search`	elastic_mcp
`auto_mapping`	elastic_mcp
`search_explain`	elastic_mcp
`list_patterns`	elastic_mcp

Ctx Elastic¶

Overview¶

Public API¶

HitRerankDecision¶

LLMHitRerankRuntime¶

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

CtxElasticRerankPatternRuntime¶

enforce_eligibility(decision: HitRerankDecision, hits: list[Any]) -> tuple[list[Any], bool, bool]¶

CtxElasticPlanner¶

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

CtxElasticBlock(AIBlock[SearchQuery, SearchResults, None])¶

infer(data: SearchQuery) -> Result[SearchResults]¶

SearchQuery(BaseModel)¶

SearchHit(BaseModel)¶

SearchResults(BaseModel)¶

CtxElasticMCPBlock(AIBlock[MCPElasticInput, MCPElasticOutput, dict])¶

infer(data: MCPElasticInput) -> Result[MCPElasticOutput]¶

MCPElasticInput(BaseModel)¶

MCPElasticRecord(BaseModel)¶

MCPElasticOutput(BaseModel)¶

ElasticStore¶

save_session(index_name: str = 'default', query_text: str = '', query_json: str = '{}', results_json: str = '[]', hit_count: int = 0, tags: list[str] | None = None, notes: str = '') -> str¶

get_sessions(index_name: str = '', tag: str = '', limit: int = 50) -> list[dict[str, Any]]¶

save_template(name: str, description: str = '', body_json: str = '{}', index_pattern: str = '', tags: list[str] | None = None) -> str¶

get_template(name: str) -> dict[str, Any] | None¶

get_templates(tag: str = '', search: str = '', limit: int = 50) -> list[dict[str, Any]]¶

save_bookmark(index_name: str = 'default', doc_id: str = '', doc_source_json: str = '{}', notes: str = '', tags: list[str] | None = None) -> str¶

get_bookmarks(index_name: str = '', tag: str = '', limit: int = 50) -> list[dict[str, Any]]¶

save_outcome(session_id: str = '', query_text: str = '', relevance_score: float = 0.0, clicked_doc_ids: list[str] | None = None, notes: str = '') -> str¶

cache_llm(prompt_text: str, response_text: str, model: str = '') -> str¶

get_cached_llm(prompt_text: str) -> str | None¶

get_config(key: str) -> str | None¶

set_config(key: str, value_json: str) -> None¶

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

Functions¶

agentic_planner_enabled(default_enabled: bool) -> bool¶

validate_hit_rerank_decision(decision: HitRerankDecision, n_eligible: int, expected_fingerprint: str) -> None¶

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

get_skill_catalog() -> CtxElasticSkillCatalog¶

MCP Tools¶

`HitRerankDecision`¶

`LLMHitRerankRuntime`¶

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

`CtxElasticRerankPatternRuntime`¶

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

`CtxElasticPlanner`¶

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

`CtxElasticBlock(AIBlock[SearchQuery, SearchResults, None])`¶

`infer(data: SearchQuery) -> Result[SearchResults]`¶

`SearchQuery(BaseModel)`¶

`SearchHit(BaseModel)`¶

`SearchResults(BaseModel)`¶

`CtxElasticMCPBlock(AIBlock[MCPElasticInput, MCPElasticOutput, dict])`¶

`infer(data: MCPElasticInput) -> Result[MCPElasticOutput]`¶

`MCPElasticInput(BaseModel)`¶

`MCPElasticRecord(BaseModel)`¶

`MCPElasticOutput(BaseModel)`¶

`ElasticStore`¶

`save_session(index_name: str = 'default', query_text: str = '', query_json: str = '{}', results_json: str = '[]', hit_count: int = 0, tags: list[str] | None = None, notes: str = '') -> str`¶

`get_sessions(index_name: str = '', tag: str = '', limit: int = 50) -> list[dict[str, Any]]`¶

`save_template(name: str, description: str = '', body_json: str = '{}', index_pattern: str = '', tags: list[str] | None = None) -> str`¶

`get_template(name: str) -> dict[str, Any] | None`¶

`get_templates(tag: str = '', search: str = '', limit: int = 50) -> list[dict[str, Any]]`¶

`save_bookmark(index_name: str = 'default', doc_id: str = '', doc_source_json: str = '{}', notes: str = '', tags: list[str] | None = None) -> str`¶

`get_bookmarks(index_name: str = '', tag: str = '', limit: int = 50) -> list[dict[str, Any]]`¶

`save_outcome(session_id: str = '', query_text: str = '', relevance_score: float = 0.0, clicked_doc_ids: list[str] | None = None, notes: str = '') -> str`¶

`cache_llm(prompt_text: str, response_text: str, model: str = '') -> str`¶

`get_cached_llm(prompt_text: str) -> str | None`¶

`get_config(key: str) -> str | None`¶

`set_config(key: str, value_json: str) -> None`¶

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

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

`validate_hit_rerank_decision(decision: HitRerankDecision, n_eligible: int, expected_fingerprint: str) -> None`¶

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

`get_skill_catalog() -> CtxElasticSkillCatalog`¶