Skip to content

Ctx Research Scanner¶

ctx_research_scanner — AI research discovery and novelty scoring.

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

Overview¶

Automated research paper discovery and triage system. Scans arXiv and custom URLs for papers relevant to G6 capability gaps, with a deterministic mock source for tests. GitHub and academic conference proceedings are visible as source capabilities but are gated as not_implemented; requests for those sources return completion_state="blocked-escalated" with warning-card/evidence metadata rather than silently scanning. Scores and ranks candidates with scoring_method="keyword_overlap" (heuristic component-name overlap, not semantic gap analysis), deduplicates across arXiv IDs, and maintains a persistent SQLite-backed session registry. Schedule settings are scheduling_mode="config_only" with active_scheduler=false.

When to use:

Running on-demand arXiv scans for new papers in a domain
Extracting arXiv paper links from a custom URL
Identifying heuristic research gaps in the current G6 component set
Building a curated reading list for a specific capability area

Example:

from mvp.ctx_research_scanner import CtxResearchScannerBlock, ScanInput

block = CtxResearchScannerBlock(name="scanner")
result = block.infer(ScanInput(source="arxiv", query="program synthesis CEGIS 2025"))
# result.ok -> True; result.value -> ScanOutput with completion_state,
# warning_card, evidence, request_id, and run_id envelope fields.

Works well with: ctx_library_mapper, adapt_library_synthesizer, ctx_scrapling

Public API¶

`CtxResearchScannerBlock(AIBlock[ScanInput, ScanOutput, None])`¶

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

Methods:

`infer(data: ScanInput) -> Result[ScanOutput]`¶

`PaperRecord(BaseModel)`¶

Field	Type	Default
`title`	`str`	required
`url`	`str`	required
`abstract`	`str`	''
`authors`	`str`	''
`date`	`str`	''
`source`	`str`	''
`arxiv_id`	`str`	''
`key_terms`	`list[str]`	Field(default_factory=list)

`ScanInput(BaseModel)`¶

Field	Type	Default
`source`	`Literal['arxiv', 'github', 'conferences', 'custom', 'mock']`	'mock'
`query`	`str`	required
`categories`	`list[str]`	Field(default_factory=list)
`max_results`	`int`	Field(default=10, ge=1, le=100)
`url`	`str`	''
`agentic_rerank`	`bool \\| None`	None
`run_mode`	`Literal['beta', 'production']`	'beta'
`reviewer_signature`	`str`	''

`ScanOutput(BaseModel)`¶

Field	Type	Default
`source`	`str`	required
`query`	`str`	required
`results`	`list[PaperRecord]`	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)

`CtxResearchScannerMCPBlock(AIBlock[MCPResearchScannerInput, MCPResearchScannerOutput, dict])`¶

25-op research scanner MCP block with SQLite persistence (+1 block-dispatch op).

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

Methods:

`store() -> ResearchScannerStore`¶

`infer(data: MCPResearchScannerInput) -> Result[MCPResearchScannerOutput]`¶

`MCPResearchScannerRecord(BaseModel)`¶

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)

`MCPResearchScannerInput(BaseModel)`¶

Field	Type	Default
`op`	`ResearchScannerOp`	required
`paper_id`	`str`	''
`title`	`str`	''
`url`	`str`	''
`abstract`	`str`	''
`arxiv_id`	`str`	''
`key_terms`	`str`	''
`seed_path`	`str`	''
`query`	`str`	''
`source`	`str`	''
`categories`	`str`	''
`max_results`	`int`	Field(default=10, ge=1, le=100)
`custom_url`	`str`	''
`session_id`	`str`	''
`domain`	`str`	''
`trigger`	`str`	''
`sources`	`str`	''
`source_id`	`str`	''
`source_type`	`str`	''
`source_url`	`str`	''
`schedule_hours`	`int`	Field(default=0, ge=0)
`limit`	`int`	Field(default=50, ge=1, le=1000)
`offset`	`int`	Field(default=0, ge=0)
`agentic_rerank`	`bool \\| None`	None
`run_mode`	`Literal['beta', 'production']`	'beta'
`reviewer_signature`	`str`	''

`MCPResearchScannerOutput(BaseModel)`¶

Field	Type	Default
`op`	`str`	''
`key`	`str`	''
`value`	`str`	''
`found`	`bool`	False
`count`	`int`	0
`records`	`list[MCPResearchScannerRecord]`	Field(default_factory=list)
`content`	`str`	''
`scores`	`list[float]`	Field(default_factory=list)
`message`	`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`	''
`agentic_evidence`	`dict[str, Any]`	Field(default_factory=dict)

`ResearchScannerStore`¶

Constructor:

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

Methods:

`add_paper(title: str, url: str = '', abstract: str = '', authors: str = '', date: str = '', source: str = '', arxiv_id: str = '', key_terms: list[str] | None = None) -> str`¶

`get_paper(paper_id: str) -> dict[str, Any] | None`¶

`remove_paper(paper_id: str) -> bool`¶

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

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

`count_papers() -> int`¶

`find_by_arxiv_id(arxiv_id: str) -> dict[str, Any] | None`¶

`start_session(trigger: str, sources: list[str]) -> str`¶

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

`finish_session(session_id: str, candidates_found: int = 0) -> None`¶

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

`add_candidate(session_id: str, title: str, url: str = '', abstract: str = '', arxiv_id: str = '', novelty_score: float = 0.0, capability_gap: str = '', domain: str = '', key_terms: list[str] | None = None) -> str`¶

`list_candidates(session_id: str = '', limit: int = 50, offset: int = 0) -> list[dict[str, Any]]`¶

`update_candidate_score(candidate_id: str, novelty_score: float, capability_gap: str = '') -> bool`¶

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

`add_source(source_type: str, url: str = '', categories: str = '') -> str`¶

`remove_source(source_id: str) -> bool`¶

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

MCP Tools¶

Operation	Source
`import_seed_papers`	research_scanner_mcp
`add_paper`	research_scanner_mcp
`remove_paper`	research_scanner_mcp
`search_papers`	research_scanner_mcp
`list_papers`	research_scanner_mcp
`scan_arxiv`	research_scanner_mcp
`scan_github`	research_scanner_mcp
`scan_conferences`	research_scanner_mcp
`scan_custom_url`	research_scanner_mcp
`scan_all`	research_scanner_mcp
`score_candidates`	research_scanner_mcp
`get_capability_gaps`	research_scanner_mcp
`rank_candidates`	research_scanner_mcp
`filter_by_domain`	research_scanner_mcp
`deduplicate`	research_scanner_mcp
`start_scan_session`	research_scanner_mcp
`get_session_status`	research_scanner_mcp
`list_sessions`	research_scanner_mcp
`get_candidates`	research_scanner_mcp
`export_candidates`	research_scanner_mcp
`add_source`	research_scanner_mcp
`remove_source`	research_scanner_mcp
`list_sources`	research_scanner_mcp
`set_schedule`	research_scanner_mcp
`research_scanner_info`	research_scanner_mcp
`list_patterns`	research_scanner_mcp