Ctx Contexthub¶

ctx_contexthub — mvp.ctx_contexthub

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

Overview¶

Curated documentation registry with BM25 search, persistent annotations, multi-source management, and caching. ContextHub ships with a packaged default bundle under components/mvp/ctx_contexthub/skill/bundles/default, so a fresh MCP extension install can run chub_update_registry, search, and fetch starter guidance without the remote CDN.

Starter bundle, not a complete knowledge base

The packaged default registry is intentionally small. It is enough for first-run checks, retrieval routing guidance, and ContextHub onboarding, but it is not a complete G6 documentation corpus. For real pilot or production workflows, add the docs your users need with chub_add_source, chub_import_local, or a maintained remote registry, then verify non-zero source and registry counts with chub_info or chub_cache_status.

When to use:

Maintaining a local searchable registry of approved documentation sources
Annotating documentation entries with tags and notes for downstream retrieval
Federating multiple doc sources (local files, URLs, Git repos) behind a unified search interface
Providing a low-friction first-run documentation source for MCP installs

Example:

from mvp.ctx_contexthub import CtxContextHubMCPBlock, MCPContextHubInput

block = CtxContextHubMCPBlock()
block.infer(MCPContextHubInput(op="update_registry"))
result = block.infer(MCPContextHubInput(op="search", query="quickstart", top_k=5))
# result.is_ok() -> True; result.value.results contains packaged starter entries

First-run check: chub_update_registry -> chub_search query="quickstart" -> chub_get entry_id="g6/contexthub-quickstart" lang="python".

Works well with: ctx_rag, ctx_elastic, ctx_recursive

Public API¶

`ContextHubRerankDecision`¶

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

`LLMContextHubRerankRuntime`¶

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

Constructor:

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

Methods:

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

`CtxContextHubRerankPatternRuntime`¶

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

Methods:

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

`CtxContextHubPlanner`¶

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

Constructor:

Parameter	Type	Default
`runtime`	`ContextHubRerankRuntime \\| None`	None
`pattern_runtime`	`CtxContextHubRerankPatternRuntime \\| None`	None

Methods:

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

`CtxContextHubBlock(AIBlock[ContextHubInput, ContextHubOutput, dict])`¶

Read-only ContextHub block: search, get, list over doc registry.

Field	Type	Default
`name`	`str`	'ctx_contexthub'
`state`	`dict \\| None`	None
`resource_bounds`	`ResourceBounds \\| None`	None
`usage`	`ResourceUsage`	field(default_factory=ResourceUsage)
`store`	`Any`	None
`agentic_planner`	`CtxContextHubPlanner \\| None`	None

Methods:

`infer(data: ContextHubInput) -> Result[ContextHubOutput]`¶

`ContextHubInput(BaseModel)`¶

Input to CtxContextHubBlock.

Field	Type	Default
`op`	`Literal['search', 'get', 'list', 'help', 'ops']`	required
`query`	`str`	''
`entry_id`	`str`	''
`lang`	`str`	''
`tags`	`str`	''
`version`	`str`	''
`full`	`bool`	False
`file`	`str`	''
`top_k`	`int`	20
`score_threshold`	`float`	0.0
`agentic_rerank`	`bool \\| None`	None
`run_mode`	`Literal['beta', 'production']`	'beta'
`reviewer_signature`	`str`	''

`ContextHubOutput(BaseModel)`¶

Output from CtxContextHubBlock.

Field	Type	Default
`op`	`str`	required
`content`	`str`	''
`results`	`list[dict[str, Any]]`	Field(default_factory=list)
`total`	`int`	0
`entry_id`	`str`	''
`entry_type`	`str`	''
`message`	`str`	''
`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`	Field(default_factory=dict)

`CtxContextHubMCPBlock(AIBlock[MCPContextHubInput, MCPContextHubOutput, dict])`¶

Full-featured ContextHub block with SQLite persistence. 25 operations

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

Methods:

`infer(data: MCPContextHubInput) -> Result[MCPContextHubOutput]`¶

`MCPContextHubRecord(BaseModel)`¶

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

`MCPContextHubInput(BaseModel)`¶

Field	Type	Default
`op`	`Literal['search', 'get', 'list', 'list_sources', 'info', 'update_registry', 'fetch_bundle', 'cache_status', 'clear_cache', 'add_source', 'annotate_write', 'annotate_read', 'annotate_clear', 'annotate_list', 'annotate_search', 'get_full', 'get_file', 'list_files', 'import_local', 'remove_local', 'feedback', 'usage_stats', 'search_history', 'popular_docs', 'quality_report', 'list_patterns', 'help', 'ops']`	required
`query`	`str`	''
`entry_id`	`str`	''
`lang`	`str`	''
`version`	`str`	''
`tags_csv`	`str`	''
`top_k`	`int`	20
`full`	`bool`	False
`score_threshold`	`float`	0.0
`source_name`	`str`	''
`source_url`	`str`	''
`source_path`	`str`	''
`note`	`str`	''
`rating`	`str`	''
`comment`	`str`	''
`labels_csv`	`str`	''
`file_path`	`str`	''
`local_dir`	`str`	''
`limit`	`int`	50
`since`	`str`	''
`agentic_rerank`	`bool \\| None`	None
`run_mode`	`Literal['beta', 'production']`	'beta'
`reviewer_signature`	`str`	''

`MCPContextHubOutput(BaseModel)`¶

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

`ContextHubStore`¶

Sync SQLite store with 7 tables.

Constructor:

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

Methods:

`add_source(name: str, url: str = '', path: str = '') -> str`¶

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

`update_source_timestamp(name: str) -> None`¶

`upsert_registry_entry(entry_id: str, source: str, name: str, entry_type: str, description: str, tags_json: str, languages_json: str, path: str = '', files_json: str = '[]') -> None`¶

`list_registry_entries(source: str = '', entry_type: str = '') -> list[dict[str, Any]]`¶

`get_registry_entries_as_dicts() -> list[dict]`¶

Return all entries as dicts with parsed JSON fields for BM25 indexing.

`clear_registry_entries(source: str = '') -> int`¶

`get_cached_doc(entry_id: str, file_path: str) -> str | None`¶

`cache_doc(entry_id: str, file_path: str, content: str, content_hash: str = '') -> None`¶

`verify_cached_doc(entry_id: str, file_path: str) -> bool | None`¶

Verify cached doc integrity. Returns True if valid, False if corrupt, None if not cached.

`write_annotation(entry_id: str, note: str) -> dict`¶

`read_annotation(entry_id: str) -> dict | None`¶

`clear_annotation(entry_id: str) -> bool`¶

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

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

`add_feedback(entry_id: str, rating: str, comment: str = '', labels_json: str = '[]') -> str`¶

`log_search(query: str, filters_json: str = '{}', result_count: int = 0) -> str`¶

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

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

`log_usage(entry_id: str, op: str) -> str`¶

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

`list_cached_files(entry_id: str) -> list[dict[str, Any]]`¶

List all cached file paths for a given entry.

`remove_source(name: str) -> int`¶

Remove a source and cascade-delete its registry entries and cached docs.

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

Aggregate quality/health stats across all tables.

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

Functions¶

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

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

`planner_is_llm_trusted(planner: Any) -> bool`¶

Whether the BLOCK may report llm_used=True for planner.

MCP Tools¶

Operation	Source
`search`	contexthub_mcp
`get`	contexthub_mcp
`list`	contexthub_mcp
`list_sources`	contexthub_mcp
`info`	contexthub_mcp
`update_registry`	contexthub_mcp
`fetch_bundle`	contexthub_mcp
`cache_status`	contexthub_mcp
`clear_cache`	contexthub_mcp
`add_source`	contexthub_mcp
`annotate_write`	contexthub_mcp
`annotate_read`	contexthub_mcp
`annotate_clear`	contexthub_mcp
`annotate_list`	contexthub_mcp
`annotate_search`	contexthub_mcp
`get_full`	contexthub_mcp
`get_file`	contexthub_mcp
`list_files`	contexthub_mcp
`import_local`	contexthub_mcp
`remove_local`	contexthub_mcp
`feedback`	contexthub_mcp
`usage_stats`	contexthub_mcp
`search_history`	contexthub_mcp
`popular_docs`	contexthub_mcp
`quality_report`	contexthub_mcp
`list_patterns`	contexthub_mcp
`help`	contexthub_mcp
`ops`	contexthub_mcp

Ctx Contexthub¶

Overview¶

Public API¶

ContextHubRerankDecision¶

LLMContextHubRerankRuntime¶

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

CtxContextHubRerankPatternRuntime¶

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

CtxContextHubPlanner¶

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

CtxContextHubBlock(AIBlock[ContextHubInput, ContextHubOutput, dict])¶

infer(data: ContextHubInput) -> Result[ContextHubOutput]¶

ContextHubInput(BaseModel)¶

ContextHubOutput(BaseModel)¶

CtxContextHubMCPBlock(AIBlock[MCPContextHubInput, MCPContextHubOutput, dict])¶

infer(data: MCPContextHubInput) -> Result[MCPContextHubOutput]¶

MCPContextHubRecord(BaseModel)¶

MCPContextHubInput(BaseModel)¶

MCPContextHubOutput(BaseModel)¶

ContextHubStore¶

add_source(name: str, url: str = '', path: str = '') -> str¶

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

update_source_timestamp(name: str) -> None¶

upsert_registry_entry(entry_id: str, source: str, name: str, entry_type: str, description: str, tags_json: str, languages_json: str, path: str = '', files_json: str = '[]') -> None¶

list_registry_entries(source: str = '', entry_type: str = '') -> list[dict[str, Any]]¶

get_registry_entries_as_dicts() -> list[dict]¶

clear_registry_entries(source: str = '') -> int¶

get_cached_doc(entry_id: str, file_path: str) -> str | None¶

cache_doc(entry_id: str, file_path: str, content: str, content_hash: str = '') -> None¶

verify_cached_doc(entry_id: str, file_path: str) -> bool | None¶

write_annotation(entry_id: str, note: str) -> dict¶

read_annotation(entry_id: str) -> dict | None¶

clear_annotation(entry_id: str) -> bool¶

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

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

add_feedback(entry_id: str, rating: str, comment: str = '', labels_json: str = '[]') -> str¶

log_search(query: str, filters_json: str = '{}', result_count: int = 0) -> str¶

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

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

log_usage(entry_id: str, op: str) -> str¶

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

list_cached_files(entry_id: str) -> list[dict[str, Any]]¶

remove_source(name: str) -> int¶

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

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

Functions¶

agentic_planner_enabled(default_enabled: bool) -> bool¶

planner_is_llm_trusted(planner: Any) -> bool¶

MCP Tools¶

`ContextHubRerankDecision`¶

`LLMContextHubRerankRuntime`¶

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

`CtxContextHubRerankPatternRuntime`¶

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

`CtxContextHubPlanner`¶

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

`CtxContextHubBlock(AIBlock[ContextHubInput, ContextHubOutput, dict])`¶

`infer(data: ContextHubInput) -> Result[ContextHubOutput]`¶

`ContextHubInput(BaseModel)`¶

`ContextHubOutput(BaseModel)`¶

`CtxContextHubMCPBlock(AIBlock[MCPContextHubInput, MCPContextHubOutput, dict])`¶

`infer(data: MCPContextHubInput) -> Result[MCPContextHubOutput]`¶

`MCPContextHubRecord(BaseModel)`¶

`MCPContextHubInput(BaseModel)`¶

`MCPContextHubOutput(BaseModel)`¶

`ContextHubStore`¶

`add_source(name: str, url: str = '', path: str = '') -> str`¶

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

`update_source_timestamp(name: str) -> None`¶

`upsert_registry_entry(entry_id: str, source: str, name: str, entry_type: str, description: str, tags_json: str, languages_json: str, path: str = '', files_json: str = '[]') -> None`¶

`list_registry_entries(source: str = '', entry_type: str = '') -> list[dict[str, Any]]`¶

`get_registry_entries_as_dicts() -> list[dict]`¶

`clear_registry_entries(source: str = '') -> int`¶

`get_cached_doc(entry_id: str, file_path: str) -> str | None`¶

`cache_doc(entry_id: str, file_path: str, content: str, content_hash: str = '') -> None`¶

`verify_cached_doc(entry_id: str, file_path: str) -> bool | None`¶

`write_annotation(entry_id: str, note: str) -> dict`¶

`read_annotation(entry_id: str) -> dict | None`¶

`clear_annotation(entry_id: str) -> bool`¶

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

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

`add_feedback(entry_id: str, rating: str, comment: str = '', labels_json: str = '[]') -> str`¶

`log_search(query: str, filters_json: str = '{}', result_count: int = 0) -> str`¶

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

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

`log_usage(entry_id: str, op: str) -> str`¶

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

`list_cached_files(entry_id: str) -> list[dict[str, Any]]`¶

`remove_source(name: str) -> int`¶

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

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

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

`planner_is_llm_trusted(planner: Any) -> bool`¶