Skip to content

Ctx Library Mapper¶

ctx_library_mapper — find implementing libraries for AI research papers.

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

Overview¶

Maps research papers and algorithm descriptions to real Python libraries. Searches PyPI, GitHub, and HuggingFace to find implementations, scores candidates by quality and licence compatibility, and maintains a curated library registry with deduplication.

Live source caveat

GitHub and HuggingFace searches use their public JSON APIs and are suitable for pilot workflows. PyPI search currently parses PyPI's public HTML search results page, so it can degrade or return fewer results if PyPI changes its markup. Treat PyPI results as candidate discovery, not a guaranteed package index. For production-critical dependency decisions, verify candidates against PyPI directly before adding them to a workflow.

Source adapters and degradation

add_source stores source configs that are executable only when their type has a registered adapter (pypi, github, or huggingface). list_sources reports adapter_status and source_capabilities; executing an unknown configured source returns completion_state: blocked-escalated with G6_E_SOURCE_ADAPTER_MISSING. Partial source failures return completion_state: qualified-draft, G6_E_SOURCE_PARTIAL_FAILURE, and per-source evidence while preserving surviving records.

Production gate

A production-mode call with a reviewer signature records verification_method: acknowledgment_only and remains qualified-draft. The signature acknowledges intended production use; it does not verify candidate relevance, license compatibility, or source currency.

When to use:

Finding existing Python libraries that implement an algorithm from a paper
Auditing a research-to-implementation pipeline for library availability
Building a curated dependency list for a new G6 component

Example:

from mvp.ctx_library_mapper import CtxLibraryMapperBlock, MapInput

block = CtxLibraryMapperBlock(name="lib_mapper")
result = block.infer(MapInput(paper_title="CEGIS synthesis", source="github"))
# result.ok → True; result.value → MapOutput with libraries ranked by score

Works well with: adapt_library_synthesizer, ctx_research_scanner, component_creator

Public API¶

`CtxLibraryMapperBlock(AIBlock[MapInput, MapOutput, None])`¶

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

Methods:

`infer(data: MapInput) -> Result[MapOutput]`¶

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

`LibraryRecord(BaseModel)`¶

Field	Type	Default
`name`	`str`	required
`url`	`str`	''
`description`	`str`	''
`source`	`str`	''
`stars`	`int`	0
`license`	`str`	''
`last_updated`	`str`	''
`relevance_score`	`float`	0.0
`review_required`	`bool`	True
`parser_path`	`str`	'unknown'
`source_freshness_seconds`	`int`	-1
`source_identifier`	`str`	''
`parse_confidence`	`float`	Field(default=0.0, ge=0.0, le=1.0)
`parse_complete`	`bool`	False
`quality_signal_coverage`	`list[str]`	Field(default_factory=list)
`degradation_metadata`	`dict`	Field(default_factory=dict)

`MapInput(BaseModel)`¶

Field	Type	Default
`paper_title`	`str`	required
`paper_url`	`str`	''
`arxiv_id`	`str`	''
`keywords`	`list[str]`	Field(default_factory=list)
`max_results`	`int`	Field(default=10, ge=1, le=50)
`source`	`Literal['pypi', 'github', 'huggingface', 'mock']`	'mock'
`agentic_rerank`	`bool \\| None`	None
`run_mode`	`Literal['beta', 'production']`	'beta'
`reviewer_signature`	`str`	''

`MapOutput(BaseModel)`¶

Field	Type	Default
`paper_title`	`str`	required
`libraries`	`list[LibraryRecord]`	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`	Field(default_factory=dict)
`evidence`	`list[dict]`	Field(default_factory=list)
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''
`verification_method`	`str`	''
`agentic_evidence`	`dict`	Field(default_factory=dict)

`CtxLibraryMapperMCPBlock(AIBlock[MCPLibraryMapperInput, MCPLibraryMapperOutput, dict])`¶

25-op library mapper MCP block with SQLite persistence (+1 block-dispatch op).

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

Methods:

`store() -> LibraryMapperStore`¶

`infer(data: MCPLibraryMapperInput) -> Result[MCPLibraryMapperOutput]`¶

`MCPLibraryMapperRecord(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)

`MCPLibraryMapperInput(BaseModel)`¶

Field	Type	Default
`op`	`LibraryMapperOp`	required
`library_id`	`str`	''
`name`	`str`	''
`url`	`str`	''
`description`	`str`	''
`source`	`str`	''
`license`	`str`	''
`stars`	`int`	0
`relevance_score`	`float`	0.0
`paper_title`	`str`	''
`paper_url`	`str`	''
`arxiv_id`	`str`	''
`keywords`	`str`	''
`max_results`	`int`	Field(default=10, ge=1, le=50)
`session_id`	`str`	''
`trigger`	`str`	''
`sources`	`str`	''
`source_id`	`str`	''
`source_type`	`str`	''
`source_url`	`str`	''
`preference_key`	`str`	''
`preference_value`	`str`	''
`quality_threshold`	`float`	0.0
`license_filter`	`str`	''
`query`	`str`	''
`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`	''

`MCPLibraryMapperOutput(BaseModel)`¶

Field	Type	Default
`op`	`str`	''
`key`	`str`	''
`value`	`str`	''
`found`	`bool`	False
`count`	`int`	0
`records`	`list[MCPLibraryMapperRecord]`	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`	`list[dict[str, Any]]`	Field(default_factory=list)
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''
`verification_method`	`str`	''
`source_capabilities`	`list[dict[str, Any]]`	Field(default_factory=list)
`scoring_basis`	`str`	''
`signal_coverage`	`list[str]`	Field(default_factory=list)
`agentic_evidence`	`dict[str, Any]`	Field(default_factory=dict)

`LibraryMapperStore`¶

Constructor:

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

Methods:

`add_library(name: str, url: str = '', description: str = '', source: str = '', stars: int = 0, license: str = '', last_updated: str = '', relevance_score: float = 0.0, quality_score: float = 0.0, keywords: list[str] | None = None) -> str`¶

`get_library(library_id: str) -> dict[str, Any] | None`¶

`remove_library(library_id: str) -> bool`¶

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

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

`count_libraries() -> int`¶

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

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

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

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

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

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

`add_result(session_id: str, library_id: str = '', library_name: str = '', library_url: str = '', relevance_score: float = 0.0, quality_score: float = 0.0, recommendation: str = '') -> str`¶

`list_results(session_id: str = '', limit: int = 50, offset: int = 0) -> 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
`add_library`	library_mapper_mcp
`remove_library`	library_mapper_mcp
`search_libraries`	library_mapper_mcp
`list_libraries`	library_mapper_mcp
`get_library`	library_mapper_mcp
`map_paper_to_libraries`	library_mapper_mcp
`search_pypi`	library_mapper_mcp
`search_github`	library_mapper_mcp
`search_huggingface`	library_mapper_mcp
`scan_all_sources`	library_mapper_mcp
`score_library`	library_mapper_mcp
`rank_libraries`	library_mapper_mcp
`filter_by_license`	library_mapper_mcp
`evaluate_quality`	library_mapper_mcp
`deduplicate`	library_mapper_mcp
`start_mapping_session`	library_mapper_mcp
`get_session_status`	library_mapper_mcp
`list_sessions`	library_mapper_mcp
`get_mapping_results`	library_mapper_mcp
`export_results`	library_mapper_mcp
`add_source`	library_mapper_mcp
`remove_source`	library_mapper_mcp
`list_sources`	library_mapper_mcp
`set_preferences`	library_mapper_mcp
`library_mapper_info`	library_mapper_mcp
`list_patterns`	library_mapper_mcp