Navigator¶

Navigator component — hierarchical MCP discovery and composition.

Cluster: Goal & Planning | Type: component | MCP Tools: 37

Overview¶

Hierarchical MCP discovery and composition engine with 36 operations spanning discovery, introspection, composition, delegation, routing, monitoring, orchestration, analytics, sessions, capability inspection, and MCP lifecycle management. Maintains a capability registry, routing engine with quality modes, circuit breaker fallback, and agency tracker that enforces progressive distillation as solutions stabilise — serving as the G6 system's central component directory.

When to use:

Dynamically discovering which G6 components or MCP tools satisfy a stated capability requirement
Composing multi-step pipelines from named components without hard-coding tool chains
Routing a request to the best available component with automatic fallback when circuit breakers trip
Inspecting maturity gates, effect signatures, circuit breakers, routing cache freshness, and MCP lifecycle state before composing or routing

Navigator scope

Navigator is a discovery, routing, and orchestration layer. It can execute through registered components, but a successful route only proves that the target component was found and invoked. Production use still depends on the target component's maturity, dependencies, input schema, tier access, and domain validation. Treat the first workflow as an MCP smoke test, not a regulated-domain assurance result.

Example:

from mvp.navigator import NavigatorMCPBlock, MCPNavigatorInput

block = NavigatorMCPBlock()
result = block.infer(MCPNavigatorInput(op="nav_search", query="semantic retrieval over documents"))
# result.ok → True; result.value → MCPNavigatorOutput with ranked component recommendations

Works well with: guide, autonomous_orchestrator, recursive_architect

Public API¶

`EffectSignature`¶

Effect grade ε = (io, cost, auth, fail).

Field	Type	Default
`io`	`bool`	False
`cost`	`float`	0.0
`auth`	`frozenset[str]`	field(default_factory=frozenset)
`fail`	`bool`	False

`CircuitBreaker`¶

Per-component circuit breaker.

Field	Type	Default
`max_failures`	`int`	3
`window_seconds`	`float`	120.0

Methods:

`is_open() -> bool`¶

True if breaker is open (too many recent failures).

`record_failure() -> None`¶

`record_success() -> None`¶

`FallbackManager`¶

Manages fallback chains and circuit breakers.

Methods:

`get_fallback_chain(component: str) -> list[str]`¶

Get fallback chain for a component (empty if unknown).

`get_next_fallback(component: str, tried: set[str] | None = None) -> str | None`¶

Get next available fallback, skipping open breakers and tried.

`execute_with_fallback(primary: str, invoke_fn: Callable[[str, str, dict[str, Any]], Result], op: str, params: dict[str, Any]) -> Result`¶

Try primary, cascade through fallbacks on failure.

`record_outcome(component: str, success: bool) -> None`¶

Record success/failure for circuit breaker tracking.

`register_chain(component: str, chain: list[str]) -> None`¶

Register a custom fallback chain.

`get_breaker_status(component: str) -> dict[str, Any]`¶

Get circuit breaker status for a component.

`NavigatorMCPBlock(AIBlock[MCPNavigatorInput, MCPNavigatorOutput, dict])`¶

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

Methods:

`infer(data: MCPNavigatorInput) -> Result[MCPNavigatorOutput]`¶

`CapabilityRegistry`¶

Registry of all MCP tools discovered via AST scanning.

Constructor:

Parameter	Type	Default
`workspace_root`	`str \\| None`	None

Methods:

`scan() -> None`¶

Scan all components for MCP tool registrations.

`reset() -> None`¶

Clear scan state so next access re-discovers all tools.

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

`search(query: str = '', cluster: str = '', component: str = '', capability: str = '', limit: int = 20) -> list[dict[str, Any]]`¶

Search tools by query, cluster, component, or capability.

`recommend(task: str, limit: int = 5) -> list[dict[str, Any]]`¶

Recommend tools for a task by content-weighted keyword matching.

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

Overall registry statistics.

`QualityMode(Enum)`¶

`ComponentScore`¶

Field	Type	Default
`component`	`str`	required
`cluster`	`str`	required
`total_score`	`float`	required
`latency_score`	`float`	required
`quality_score`	`float`	required
`competence_score`	`float`	required
`health_score`	`float`	required

`RoutingRequest`¶

Field	Type	Default
`task`	`str`	required
`quality_mode`	`QualityMode`	QualityMode.BALANCED
`required_clusters`	`list[str]`	field(default_factory=list)
`excluded_components`	`list[str]`	field(default_factory=list)
`max_candidates`	`int`	5

`RoutingEngine`¶

Scores and ranks components for a given task.

Field	Type	Default
`ttl_seconds`	`float`	_DEFAULT_TTL

Methods:

`route(request: RoutingRequest) -> list[ComponentScore]`¶

Score and rank components for a task.

`update_health(component: str, score: float) -> None`¶

Update health score for a component (0.0-1.0).

`update_competence(component: str, score: float) -> None`¶

Update competence score for a component (0.0-1.0).

`update_latency(component: str, latency_ms: float) -> None`¶

Update observed latency for a component.

`update_quality(component: str, score: float) -> None`¶

Update quality score for a component (0.0-1.0).

`MCPNavigatorInput(BaseModel)`¶

Field	Type	Default
`op`	`NavOp`	required
`query`	`str`	''
`component`	`str`	''
`tool_name`	`str`	''
`tool_name_b`	`str`	''
`cluster`	`str`	''
`capability`	`str`	''
`mcp_name`	`str`	''
`tier`	`str`	''
`tags_json`	`str`	'[]'
`steps_json`	`str`	'[]'
`pipeline_id`	`str`	''
`pipeline_name`	`str`	''
`delegate_op`	`str`	''
`delegate_args_json`	`str`	'{}'
`delegations_json`	`str`	'[]'
`code`	`str`	''
`error_message`	`str`	''
`prompt`	`str`	''
`allowed_tools`	`str`	'Read,Grep,Glob'
`output_format`	`str`	'json'
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''
`max_agency`	`int`	5
`effect_budget_json`	`str`	'{}'
`limit`	`int`	20
`quality_mode`	`str`	'balanced'
`task_description`	`str`	''
`components`	`list[str]`	Field(default_factory=list)

`MCPNavigatorOutput(BaseModel)`¶

Field	Type	Default
`op`	`str`	required
`key`	`str`	''
`value`	`Any`	None
`found`	`bool`	False
`count`	`int`	0
`records`	`list[dict[str, Any]]`	Field(default_factory=list)
`retrieved`	`list[str]`	Field(default_factory=list)
`scores`	`list[float]`	Field(default_factory=list)
`summary`	`str`	''
`message`	`str`	''
`metadata`	`dict[str, Any]`	Field(default_factory=dict)
`degraded`	`bool`	False
`degradation_reason`	`str \\| None`	None
`completion_state`	`Literal['verified', 'qualified-draft', 'blocked-escalated']`	'qualified-draft'
`warning_card`	`dict[str, Any] \\| None`	None
`evidence`	`dict[str, Any]`	Field(default_factory=dict)
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''

`NavigatorStore`¶

Constructor:

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

Methods:

`cache_registry(tool_key: str, entry: dict[str, Any]) -> str`¶

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

`save_pipeline(name: str, steps: list[dict[str, Any]], validation: dict[str, Any] | None = None, effect_budget: dict[str, Any] | None = None) -> str`¶

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

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

`record_delegation(component: str, op: str, args: dict[str, Any] | None = None, result_data: dict[str, Any] | None = None, success: bool = True, agency_level: int = 2) -> str`¶

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

`log_distillation(pipeline_id: str, original_agency: int, distilled_agency: int, savings: dict[str, Any]) -> str`¶

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

`record_routing(task: str, quality_mode: str, selected_component: str, score: float, success: bool = True, latency_ms: float = 0.0) -> str`¶

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

`update_circuit_breaker(component: str, failure_count: int, is_open: bool) -> str`¶

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

`record_mcp_load(mcp_name: str, status: str = 'loaded') -> str`¶

`record_mcp_unload(mcp_name: str, reason: str = 'explicit') -> str`¶

`record_mcp_access(mcp_name: str) -> str`¶

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

Functions¶

`suggest_distillation(pipeline_steps: list[dict[str, Any]]) -> list[dict[str, Any]]`¶

Analyze each step and suggest lower-agency alternatives.

`enforce_max_agency(step: dict[str, Any], max_level: AgencyLevel) -> RResult[dict[str, Any], str]`¶

Reject steps exceeding the allowed agency level.

`compose_effects(e1: EffectSignature, e2: EffectSignature) -> EffectSignature`¶

Effect composition ε₁⊕ε₂ (Definition 2.7).

`check_budget(composed: EffectSignature, budget: EffectSignature) -> RResult[bool, str]`¶

Verify ε ≤ₑ ε_max (Definition 2.6 ordering).

MCP Tools¶

Operation	Source
`nav_discover`	navigator_mcp
`nav_search`	navigator_mcp
`nav_recommend`	navigator_mcp
`nav_discover_by_capability`	navigator_mcp
`nav_discover_by_protocol`	navigator_mcp
`nav_inspect_mcp`	navigator_mcp
`nav_examine_tool`	navigator_mcp
`nav_diff_tools`	navigator_mcp
`nav_inspect_agent`	navigator_mcp
`nav_compose`	navigator_mcp
`nav_validate_pipeline`	navigator_mcp
`nav_execute_plan`	navigator_mcp
`nav_delegate`	navigator_mcp
`nav_batch_delegate`	navigator_mcp
`nav_self_heal`	navigator_mcp
`nav_distill`	navigator_mcp
`nav_claude_headless`	navigator_mcp
`nav_stats`	navigator_mcp
`nav_inspect_capabilities`	navigator_mcp
`nav_unified_search`	navigator_mcp
`nav_health_check`	navigator_mcp
`nav_dependency_graph`	navigator_mcp
`nav_execute_workflow`	navigator_mcp
`nav_get_execution`	navigator_mcp
`nav_list_executions`	navigator_mcp
`nav_aggregate_stats`	navigator_mcp
`nav_cross_reference`	navigator_mcp
`nav_save_session`	navigator_mcp
`nav_load_session`	navigator_mcp
`nav_route`	navigator_mcp
`nav_refresh_catalog`	navigator_mcp
`nav_mcp_load`	navigator_mcp
`nav_mcp_unload`	navigator_mcp
`nav_mcp_list_active`	navigator_mcp
`nav_mcp_evict_idle`	navigator_mcp
`nav_mcp_registry`	navigator_mcp
`list_patterns`	navigator_mcp

Dependencies¶

mvp.core

Navigator¶

Overview¶

Public API¶

EffectSignature¶

CircuitBreaker¶

is_open() -> bool¶

record_failure() -> None¶

record_success() -> None¶

FallbackManager¶

get_fallback_chain(component: str) -> list[str]¶

get_next_fallback(component: str, tried: set[str] | None = None) -> str | None¶

execute_with_fallback(primary: str, invoke_fn: Callable[[str, str, dict[str, Any]], Result], op: str, params: dict[str, Any]) -> Result¶

record_outcome(component: str, success: bool) -> None¶

register_chain(component: str, chain: list[str]) -> None¶

get_breaker_status(component: str) -> dict[str, Any]¶

NavigatorMCPBlock(AIBlock[MCPNavigatorInput, MCPNavigatorOutput, dict])¶

infer(data: MCPNavigatorInput) -> Result[MCPNavigatorOutput]¶

CapabilityRegistry¶

scan() -> None¶

reset() -> None¶

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

search(query: str = '', cluster: str = '', component: str = '', capability: str = '', limit: int = 20) -> list[dict[str, Any]]¶

recommend(task: str, limit: int = 5) -> list[dict[str, Any]]¶

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

QualityMode(Enum)¶

ComponentScore¶

RoutingRequest¶

RoutingEngine¶

route(request: RoutingRequest) -> list[ComponentScore]¶

update_health(component: str, score: float) -> None¶

update_competence(component: str, score: float) -> None¶

update_latency(component: str, latency_ms: float) -> None¶

update_quality(component: str, score: float) -> None¶

MCPNavigatorInput(BaseModel)¶

MCPNavigatorOutput(BaseModel)¶

NavigatorStore¶

cache_registry(tool_key: str, entry: dict[str, Any]) -> str¶

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

save_pipeline(name: str, steps: list[dict[str, Any]], validation: dict[str, Any] | None = None, effect_budget: dict[str, Any] | None = None) -> str¶

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

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

record_delegation(component: str, op: str, args: dict[str, Any] | None = None, result_data: dict[str, Any] | None = None, success: bool = True, agency_level: int = 2) -> str¶

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

log_distillation(pipeline_id: str, original_agency: int, distilled_agency: int, savings: dict[str, Any]) -> str¶

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

record_routing(task: str, quality_mode: str, selected_component: str, score: float, success: bool = True, latency_ms: float = 0.0) -> str¶

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

update_circuit_breaker(component: str, failure_count: int, is_open: bool) -> str¶

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

record_mcp_load(mcp_name: str, status: str = 'loaded') -> str¶

record_mcp_unload(mcp_name: str, reason: str = 'explicit') -> str¶

record_mcp_access(mcp_name: str) -> str¶

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

Functions¶

suggest_distillation(pipeline_steps: list[dict[str, Any]]) -> list[dict[str, Any]]¶

enforce_max_agency(step: dict[str, Any], max_level: AgencyLevel) -> RResult[dict[str, Any], str]¶

compose_effects(e1: EffectSignature, e2: EffectSignature) -> EffectSignature¶

check_budget(composed: EffectSignature, budget: EffectSignature) -> RResult[bool, str]¶

MCP Tools¶

Dependencies¶

`EffectSignature`¶

`CircuitBreaker`¶

`is_open() -> bool`¶

`record_failure() -> None`¶

`record_success() -> None`¶

`FallbackManager`¶

`get_fallback_chain(component: str) -> list[str]`¶

`get_next_fallback(component: str, tried: set[str] | None = None) -> str | None`¶

`execute_with_fallback(primary: str, invoke_fn: Callable[[str, str, dict[str, Any]], Result], op: str, params: dict[str, Any]) -> Result`¶

`record_outcome(component: str, success: bool) -> None`¶

`register_chain(component: str, chain: list[str]) -> None`¶

`get_breaker_status(component: str) -> dict[str, Any]`¶

`NavigatorMCPBlock(AIBlock[MCPNavigatorInput, MCPNavigatorOutput, dict])`¶

`infer(data: MCPNavigatorInput) -> Result[MCPNavigatorOutput]`¶

`CapabilityRegistry`¶

`scan() -> None`¶

`reset() -> None`¶

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

`search(query: str = '', cluster: str = '', component: str = '', capability: str = '', limit: int = 20) -> list[dict[str, Any]]`¶

`recommend(task: str, limit: int = 5) -> list[dict[str, Any]]`¶

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

`QualityMode(Enum)`¶

`ComponentScore`¶

`RoutingRequest`¶

`RoutingEngine`¶

`route(request: RoutingRequest) -> list[ComponentScore]`¶

`update_health(component: str, score: float) -> None`¶

`update_competence(component: str, score: float) -> None`¶

`update_latency(component: str, latency_ms: float) -> None`¶

`update_quality(component: str, score: float) -> None`¶

`MCPNavigatorInput(BaseModel)`¶

`MCPNavigatorOutput(BaseModel)`¶

`NavigatorStore`¶

`cache_registry(tool_key: str, entry: dict[str, Any]) -> str`¶

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

`save_pipeline(name: str, steps: list[dict[str, Any]], validation: dict[str, Any] | None = None, effect_budget: dict[str, Any] | None = None) -> str`¶

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

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

`record_delegation(component: str, op: str, args: dict[str, Any] | None = None, result_data: dict[str, Any] | None = None, success: bool = True, agency_level: int = 2) -> str`¶

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

`log_distillation(pipeline_id: str, original_agency: int, distilled_agency: int, savings: dict[str, Any]) -> str`¶

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

`record_routing(task: str, quality_mode: str, selected_component: str, score: float, success: bool = True, latency_ms: float = 0.0) -> str`¶

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

`update_circuit_breaker(component: str, failure_count: int, is_open: bool) -> str`¶

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

`record_mcp_load(mcp_name: str, status: str = 'loaded') -> str`¶

`record_mcp_unload(mcp_name: str, reason: str = 'explicit') -> str`¶

`record_mcp_access(mcp_name: str) -> str`¶

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

`suggest_distillation(pipeline_steps: list[dict[str, Any]]) -> list[dict[str, Any]]`¶

`enforce_max_agency(step: dict[str, Any], max_level: AgencyLevel) -> RResult[dict[str, Any], str]`¶

`compose_effects(e1: EffectSignature, e2: EffectSignature) -> EffectSignature`¶

`check_budget(composed: EffectSignature, budget: EffectSignature) -> RResult[bool, str]`¶