Ctx Search¶

ctx_search — mvp.ctx_search

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

Overview¶

Web search abstraction supporting core mock, DuckDuckGo, and Brave backends plus an MCP research surface with mock, DuckDuckGo, SearXNG, academic, and social wrappers. Outputs carry normalized source fields and canonical degradation envelopes so callers can distinguish true no-hit results from backend, network, credential, or constraint failures.

Launch smoke-test caveat

ctx_search has passing component and MCP regression coverage, and the MCP-facing search_web path defaults to live DuckDuckGo rather than canned mock results. The full tests/mvp/ctx_search/test_search_mcp.py suite is intentionally broad and can take about two minutes on a development machine, so do not use it as the first-user onboarding proof. Before a pilot or launch demo, run a separate clean-install MCP smoke test that starts the server, calls search_info, runs one live search_web query, then runs a small run_research query and verifies that the returned research_id can be used with research_status.

Backend and credential gating

Brave is core-only and requires BRAVE_SEARCH_API_KEY plus paid-API opt-in. MCP SearXNG/arXiv/PubMed/Reddit paths require SEARCH_MCP_ENABLE_NETWORK=1 and report qualified-draft when disabled or unavailable. Capability discovery reports credential presence as booleans only; API key values are never surfaced.

When to use:

Retrieving live web results to augment reasoning
Augmenting LLM context with current information
Building multi-step research pipelines

Example:

from mvp.ctx_search import CtxSearchBlock, SearchInput

block = CtxSearchBlock(name="s")
result = block.infer(SearchInput(query="latest Python release", engine="duckduckgo"))
# result.ok → True; result.value → SearchOutput with web results

Works well with: ctx_rag, grounding, ctx_scrapling

Public API¶

`SearchRerankDecision`¶

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

`LLMSearchRerankRuntime`¶

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

Constructor:

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

Methods:

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

`CtxSearchRerankPatternRuntime`¶

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

Methods:

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

`CtxSearchPlanner`¶

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

Constructor:

Parameter	Type	Default
`runtime`	`SearchRerankRuntime \\| None`	None
`pattern_runtime`	`CtxSearchRerankPatternRuntime \\| None`	None

Methods:

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

`WebResult(BaseModel)`¶

A single web search result.

Field	Type	Default
`title`	`str`	required
`url`	`str`	required
`snippet`	`str`	''
`content`	`str`	''
`backend`	`str`	''
`source_type`	`str`	'web'
`query`	`str`	''
`retrieval_timestamp`	`str`	''
`fetch_status`	`str`	'unknown'
`freshness`	`str`	'unknown'
`authority_score`	`float \\| None`	None
`license_caveat`	`str`	''
`degradation_metadata`	`dict[str, Any]`	Field(default_factory=dict)

`SearchInput(BaseModel)`¶

Input to CtxSearchBlock.

Field	Type	Default
`query`	`str`	required
`engine`	`Literal['mock', 'duckduckgo', 'brave']`	'duckduckgo'
`num_results`	`int`	5
`fetch_content`	`bool`	False
`allow_paid_api`	`bool`	False
`agentic_rerank`	`bool \\| None`	None
`run_mode`	`Literal['beta', 'production']`	'beta'
`reviewer_signature`	`str`	''
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''

`SearchOutput(BaseModel)`¶

Output from CtxSearchBlock.

Field	Type	Default
`query`	`str`	required
`results`	`list[WebResult]`	required
`engine`	`str`	required
`total_found`	`int`	required
`degraded`	`bool`	False
`degradation_reason`	`str \\| None`	None
`completion_state`	`Literal['verified', 'qualified-draft', 'blocked-escalated']`	'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`	Field(default_factory=dict)

`CtxSearchBlock(AIBlock[SearchInput, SearchOutput, None])`¶

Web search adapter.

Field	Type	Default
`name`	`str`	'ctx_search'
`resource_bounds`	`ResourceBounds \\| None`	None
`usage`	`ResourceUsage`	field(default_factory=ResourceUsage)
`agentic_planner`	`CtxSearchPlanner \\| None`	None

Methods:

`infer(data: SearchInput) -> Result[SearchOutput]`¶

`egress_plan(op: str, params: dict) -> 'list[EgressTarget] | None'`¶

`MCPSearchRecord(BaseModel)`¶

A single stored record.

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)
`backend`	`str`	''
`source_type`	`str`	''
`query`	`str`	''
`retrieval_timestamp`	`str`	''
`fetch_status`	`str`	'unknown'
`freshness`	`str`	'unknown'
`authority_score`	`float \\| None`	None
`license_caveat`	`str`	''
`degradation_metadata`	`dict[str, Any]`	Field(default_factory=dict)

`MCPSearchInput(BaseModel)`¶

Input for any of the 33 search MCP operations.

Field	Type	Default
`op`	`SearchOp`	required
`query`	`str`	''
`engine`	`str`	'duckduckgo'
`num_results`	`int`	5
`queries_json`	`str`	''
`key`	`str`	''
`value`	`str`	''
`url`	`str`	''
`title`	`str`	''
`snippet`	`str`	''
`content`	`str`	''
`results_json`	`str`	''
`allowed_domains_json`	`str`	''
`blocked_domains_json`	`str`	''
`relevance_threshold`	`float`	0.2
`session_name`	`str`	''
`session_id`	`str`	''
`depth`	`str`	'balanced'
`urls_json`	`str`	''
`research_id`	`str`	''
`config_key`	`str`	''
`config_value`	`str`	''
`tags`	`str`	''
`limit`	`int`	50
`top_k`	`int`	5
`agentic_rerank`	`bool \\| None`	None
`run_mode`	`Literal['beta', 'production']`	'beta'
`reviewer_signature`	`str`	''
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''

`MCPSearchOutput(BaseModel)`¶

Output from any of the 33 search MCP operations.

Field	Type	Default
`op`	`str`	''
`key`	`str`	''
`value`	`str`	''
`found`	`bool`	False
`count`	`int`	0
`records`	`list[MCPSearchRecord]`	Field(default_factory=list)
`retrieved`	`list[str]`	Field(default_factory=list)
`scores`	`list[float]`	Field(default_factory=list)
`summary`	`str`	''
`message`	`str`	''
`results_json`	`str`	''
`metadata`	`dict[str, Any]`	Field(default_factory=dict)
`degraded`	`bool`	False
`degradation_reason`	`str`	''
`completion_state`	`Literal['verified', 'qualified-draft', 'blocked-escalated']`	'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`	''
`llm_used`	`bool`	False
`llm_parse_failure`	`bool`	False
`rule_fallback_triggered`	`bool`	False
`source_freshness_max_age_days`	`int`	0
`quality_caveats`	`list[str]`	Field(default_factory=list)
`agentic_evidence`	`dict[str, Any]`	Field(default_factory=dict)

`CtxSearchMCPBlock(AIBlock[MCPSearchInput, MCPSearchOutput, dict])`¶

33-op search MCP block with SQLite persistence (+1 block-dispatch

Field	Type	Default
`name`	`str`	'ctx_search_mcp'
`resource_bounds`	`ResourceBounds \\| None`	None
`usage`	`ResourceUsage`	field(default_factory=ResourceUsage)
`state`	`dict`	field(default_factory=dict)
`db_path`	`str`	':memory:'
`agentic_planner`	`CtxSearchPlanner \\| None`	None

Methods:

`store() -> SearchStore`¶

`infer(data: MCPSearchInput) -> Result[MCPSearchOutput]`¶

`SearchStore`¶

SQLite-backed store for search results, sessions, domain lists, and config.

Constructor:

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

Methods:

`add_result(key: str, url: str = '', title: str = '', snippet: str = '', content: str = '', tags: list[str] | None = None, engine: str = 'mock', score: float = 0.0) -> str`¶

`get_result(key: str) -> dict[str, Any] | None`¶

`list_results(engine: str = '', tags: list[str] | None = None, limit: int = 50) -> list[dict[str, Any]]`¶

`delete_result(key: str) -> bool`¶

`tag_result(key: str, tags: list[str]) -> bool`¶

`text_search(query: str, limit: int = 20) -> list[dict[str, Any]]`¶

Simple substring search across title, snippet, content.

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

`create_session(name: str, query: str = '', strategy: str = '', plan_json: dict[str, Any] | None = None) -> str`¶

`get_session(session_id: str) -> dict[str, Any] | None`¶

`update_session_status(session_id: str, status: str) -> bool`¶

`link_result_to_session(session_id: str, result_id: str) -> None`¶

`get_session_results(session_id: str) -> list[dict[str, Any]]`¶

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

`add_domain(domain: str, list_type: str = 'allow', reason: str = '') -> str`¶

`get_domains(list_type: str = '') -> list[dict[str, Any]]`¶

`is_domain_blocked(url: str) -> bool`¶

Check if domain from URL is in block list.

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

`get_config(key: str, default: str = '') -> str`¶

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

`create_research_run(query: str, depth: str = 'balanced', session_id: str = '', classification: dict[str, Any] | None = None) -> str`¶

Create a new research run and return its ID.

`update_research_run(run_id: str, status: str | None = None, iterations: int | None = None, results_json: str | None = None, action_log: str | None = None, summary: str | None = None) -> bool`¶

Update fields on a research run. Only non-None values are set.

`get_research_run(run_id: str) -> dict[str, Any] | None`¶

Get a research run by ID.

`list_research_runs(limit: int = 50) -> list[dict[str, Any]]`¶

List recent research runs.

Functions¶

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

Decide whether the agentic search-result-rerank planner should be used.

`eligible_fingerprint(results: list[Any]) -> str`¶

sha256 over the returned result texts (order-sensitive).

`validate_search_rerank_decision(decision: SearchRerankDecision, n_eligible: int, expected_fingerprint: str) -> None`¶

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

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

Return compact metadata for ctx_search-applied vendored patterns.

`get_skill_catalog() -> CtxSearchSkillCatalog`¶

MCP Tools¶

Operation	Source
`search`	search_mcp
`search_batch`	search_mcp
`search_engines`	search_mcp
`rank_results`	search_mcp
`diversify_results`	search_mcp
`store_result`	search_mcp
`retrieve_result`	search_mcp
`list_results`	search_mcp
`delete_result`	search_mcp
`search_stored`	search_mcp
`tag_result`	search_mcp
`result_stats`	search_mcp
`curate`	search_mcp
`vet_source`	search_mcp
`deduplicate`	search_mcp
`filter_relevance`	search_mcp
`redact_pii`	search_mcp
`score_quality`	search_mcp
`decompose_query`	search_mcp
`plan_research`	search_mcp
`create_session`	search_mcp
`session_status`	search_mcp
`end_session`	search_mcp
`classify_query`	search_mcp
`research`	search_mcp
`research_status`	search_mcp
`research_results`	search_mcp
`academic_search`	search_mcp
`social_search`	search_mcp
`scrape_urls`	search_mcp
`research_summary`	search_mcp
`search_info`	search_mcp
`configure`	search_mcp
`list_patterns`	search_mcp

Ctx Search¶

Overview¶

Public API¶

SearchRerankDecision¶

LLMSearchRerankRuntime¶

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

CtxSearchRerankPatternRuntime¶

enforce_eligibility(decision: SearchRerankDecision, results: list[Any]) -> tuple[list[Any], bool, bool]¶

CtxSearchPlanner¶

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

WebResult(BaseModel)¶

SearchInput(BaseModel)¶

SearchOutput(BaseModel)¶

CtxSearchBlock(AIBlock[SearchInput, SearchOutput, None])¶

infer(data: SearchInput) -> Result[SearchOutput]¶

egress_plan(op: str, params: dict) -> 'list[EgressTarget] | None'¶

MCPSearchRecord(BaseModel)¶

MCPSearchInput(BaseModel)¶

MCPSearchOutput(BaseModel)¶

CtxSearchMCPBlock(AIBlock[MCPSearchInput, MCPSearchOutput, dict])¶

store() -> SearchStore¶

infer(data: MCPSearchInput) -> Result[MCPSearchOutput]¶

SearchStore¶

add_result(key: str, url: str = '', title: str = '', snippet: str = '', content: str = '', tags: list[str] | None = None, engine: str = 'mock', score: float = 0.0) -> str¶

get_result(key: str) -> dict[str, Any] | None¶

list_results(engine: str = '', tags: list[str] | None = None, limit: int = 50) -> list[dict[str, Any]]¶

delete_result(key: str) -> bool¶

tag_result(key: str, tags: list[str]) -> bool¶

text_search(query: str, limit: int = 20) -> list[dict[str, Any]]¶

result_stats() -> dict[str, Any]¶

create_session(name: str, query: str = '', strategy: str = '', plan_json: dict[str, Any] | None = None) -> str¶

get_session(session_id: str) -> dict[str, Any] | None¶

update_session_status(session_id: str, status: str) -> bool¶

link_result_to_session(session_id: str, result_id: str) -> None¶

get_session_results(session_id: str) -> list[dict[str, Any]]¶

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

add_domain(domain: str, list_type: str = 'allow', reason: str = '') -> str¶

get_domains(list_type: str = '') -> list[dict[str, Any]]¶

is_domain_blocked(url: str) -> bool¶

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

get_config(key: str, default: str = '') -> str¶

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

create_research_run(query: str, depth: str = 'balanced', session_id: str = '', classification: dict[str, Any] | None = None) -> str¶

update_research_run(run_id: str, status: str | None = None, iterations: int | None = None, results_json: str | None = None, action_log: str | None = None, summary: str | None = None) -> bool¶

get_research_run(run_id: str) -> dict[str, Any] | None¶

list_research_runs(limit: int = 50) -> list[dict[str, Any]]¶

Functions¶

agentic_planner_enabled(default_enabled: bool) -> bool¶

eligible_fingerprint(results: list[Any]) -> str¶

validate_search_rerank_decision(decision: SearchRerankDecision, n_eligible: int, expected_fingerprint: str) -> None¶

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

get_skill_catalog() -> CtxSearchSkillCatalog¶

MCP Tools¶

`SearchRerankDecision`¶

`LLMSearchRerankRuntime`¶

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

`CtxSearchRerankPatternRuntime`¶

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

`CtxSearchPlanner`¶

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

`WebResult(BaseModel)`¶

`SearchInput(BaseModel)`¶

`SearchOutput(BaseModel)`¶

`CtxSearchBlock(AIBlock[SearchInput, SearchOutput, None])`¶

`infer(data: SearchInput) -> Result[SearchOutput]`¶

`egress_plan(op: str, params: dict) -> 'list[EgressTarget] | None'`¶

`MCPSearchRecord(BaseModel)`¶

`MCPSearchInput(BaseModel)`¶

`MCPSearchOutput(BaseModel)`¶

`CtxSearchMCPBlock(AIBlock[MCPSearchInput, MCPSearchOutput, dict])`¶

`store() -> SearchStore`¶

`infer(data: MCPSearchInput) -> Result[MCPSearchOutput]`¶

`SearchStore`¶

`add_result(key: str, url: str = '', title: str = '', snippet: str = '', content: str = '', tags: list[str] | None = None, engine: str = 'mock', score: float = 0.0) -> str`¶

`get_result(key: str) -> dict[str, Any] | None`¶

`list_results(engine: str = '', tags: list[str] | None = None, limit: int = 50) -> list[dict[str, Any]]`¶

`delete_result(key: str) -> bool`¶

`tag_result(key: str, tags: list[str]) -> bool`¶

`text_search(query: str, limit: int = 20) -> list[dict[str, Any]]`¶

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

`create_session(name: str, query: str = '', strategy: str = '', plan_json: dict[str, Any] | None = None) -> str`¶

`get_session(session_id: str) -> dict[str, Any] | None`¶

`update_session_status(session_id: str, status: str) -> bool`¶

`link_result_to_session(session_id: str, result_id: str) -> None`¶

`get_session_results(session_id: str) -> list[dict[str, Any]]`¶

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

`add_domain(domain: str, list_type: str = 'allow', reason: str = '') -> str`¶

`get_domains(list_type: str = '') -> list[dict[str, Any]]`¶

`is_domain_blocked(url: str) -> bool`¶

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

`get_config(key: str, default: str = '') -> str`¶

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

`create_research_run(query: str, depth: str = 'balanced', session_id: str = '', classification: dict[str, Any] | None = None) -> str`¶

`update_research_run(run_id: str, status: str | None = None, iterations: int | None = None, results_json: str | None = None, action_log: str | None = None, summary: str | None = None) -> bool`¶

`get_research_run(run_id: str) -> dict[str, Any] | None`¶

`list_research_runs(limit: int = 50) -> list[dict[str, Any]]`¶

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

`eligible_fingerprint(results: list[Any]) -> str`¶

`validate_search_rerank_decision(decision: SearchRerankDecision, n_eligible: int, expected_fingerprint: str) -> None`¶

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

`get_skill_catalog() -> CtxSearchSkillCatalog`¶