Cognitive Router¶

The cognitive router compiles natural-language tasks onto proven orchestration shapes. The LLM fills one typed GoalFrame; everything after that — protocol selection, policy gating, compilation — is rule-based.

Dispatch¶

Router ¶

Router(*, extractor: Agent, compiler: CognitiveCompiler, on_frame: Callable[[GoalFrame], None] | None = None)

One-shot dispatcher: text in, :class:RunnableResult out.

Parameters¶

extractor: :class:~locus.Agent configured with output_schema=GoalFrame. Its parsed result drives selection. compiler: :class:CognitiveCompiler set up with protocols, capabilities, policy, and model. on_frame: Optional callback fired right after a frame is extracted. Useful for telemetry / workbench display.

Source code in src/locus/router/runtime.py

def __init__(
    self,
    *,
    extractor: Agent,
    compiler: CognitiveCompiler,
    on_frame: Callable[[GoalFrame], None] | None = None,
) -> None:
    self.extractor = extractor
    self.compiler = compiler
    self._on_frame = on_frame

extract `async` ¶

extract(user_input: str, run_id: str | None = None) -> GoalFrame

Run the extractor and return the parsed :class:GoalFrame.

Emits router.frame.extracted on success or router.frame.failed on schema rejection, scoped to run_id when supplied. Raises :class:FrameExtractionError on failure — the compiler can't recover from that.

Source code in src/locus/router/runtime.py

async def extract(self, user_input: str, run_id: str | None = None) -> GoalFrame:
    """Run the extractor and return the parsed :class:`GoalFrame`.

    Emits ``router.frame.extracted`` on success or
    ``router.frame.failed`` on schema rejection, scoped to
    ``run_id`` when supplied. Raises :class:`FrameExtractionError`
    on failure — the compiler can't recover from that.
    """
    result: Any = await asyncio.to_thread(self.extractor.invoke, user_input)
    parsed = result.parsed
    if not isinstance(parsed, GoalFrame):
        err_msg = (
            "Extractor did not produce a GoalFrame. "
            f"parse_error={result.parse_error!r}, message={result.message!r}"
        )
        if run_id:
            await emit_frame_failed(run_id, err_msg)
        raise FrameExtractionError(err_msg)
    if run_id:
        await emit_frame_extracted(run_id, parsed)
    if self._on_frame is not None:
        self._on_frame(parsed)
    return parsed

dispatch `async` ¶

dispatch(user_input: str, run_id: str | None = None) -> RunnableResult

Extract a frame, compile a runnable, execute it.

run_id scopes every emitted :class:StreamEvent to one cognitive dispatch. Defaults to a fresh uuid4 if omitted — callers that want their own correlation id (a request id, a Slack thread, etc.) should pass one in. The id is also attached to the :class:RunnableResult raw payload so downstream callers can correlate without re-parsing the bus.

Source code in src/locus/router/runtime.py

async def dispatch(self, user_input: str, run_id: str | None = None) -> RunnableResult:
    """Extract a frame, compile a runnable, execute it.

    ``run_id`` scopes every emitted :class:`StreamEvent` to one
    cognitive dispatch. Defaults to a fresh ``uuid4`` if omitted —
    callers that want their own correlation id (a request id, a
    Slack thread, etc.) should pass one in. The id is also
    attached to the :class:`RunnableResult` raw payload so
    downstream callers can correlate without re-parsing the bus.
    """
    rid = run_id or str(uuid4())
    # Pin the run_id on the current asyncio context so any
    # nested instrumentation (orchestrators, specialists,
    # pipelines, skills) auto-tags its events without us
    # threading the id through every signature.
    token = set_run_id(rid)
    try:
        frame = await self.extract(user_input, run_id=rid)
        runnable = await self.compiler.compile(frame, run_id=rid)
        protocol_id = getattr(runnable, "protocol_id", "?")
        inner = getattr(runnable, "inner", None)
        if protocol_id == "?" and inner is not None:
            protocol_id = getattr(inner, "protocol_id", "?")
        await emit_runnable_executing(rid, protocol_id)
        try:
            result = await runnable.execute(user_input)
        except Exception as exc:  # noqa: BLE001 — surface every failure on bus
            await emit_runnable_failed(rid, protocol_id, f"{type(exc).__name__}: {exc}")
            raise
        await emit_runnable_executed(rid, protocol_id, len(result.text or ""))
        return result
    finally:
        # Drain history and signal end-of-stream to subscribers so
        # the SSE consumer sees a clean termination, not a hung
        # connection. Then unbind the contextvar.
        await get_event_bus().close_stream(rid)
        reset_run_id(token)

FrameExtractionError ¶

Bases: RuntimeError

Raised when the extractor agent failed to produce a valid GoalFrame.

RunnableResult ¶

Bases: BaseModel

Normalized result envelope for any compiled router execution.

Goal Frame¶

The typed contract the LLM fills. Picker / compiler read this structure deterministically; the LLM never names a protocol directly.

GoalFrame ¶

Bases: BaseModel

The single typed object the LLM produces.

Everything downstream of GoalFrame — protocol selection, capability binding, policy verdict, builder dispatch — is deterministic. The LLM never authors graph topology, only this schema.

TaskType ¶

Bases: StrEnum

The kind of cognitive work the user is asking for.

Every protocol declares which TaskType values it can handle, so the registry can filter candidates by frame.primary_goal alone.

Risk ¶

Bases: _OrderedStrEnum

Complexity ¶

Bases: _OrderedStrEnum

Protocol registry¶

A protocol is a named, parameterized orchestration shape (e.g. "specialist_fanout", "deep_research", "human_review"). Built-ins are in builtin_protocols(); users register custom shapes via ProtocolRegistry.register().

Protocol ¶

Bases: BaseModel

Declarative description of an orchestration shape.

ProtocolRegistry ¶

ProtocolRegistry()

Deterministic mapping from :class:GoalFrame to :class:Protocol.

Source code in src/locus/router/protocol.py

def __init__(self) -> None:
    self._protocols: dict[str, Protocol] = {}

filter_candidates ¶

filter_candidates(frame: GoalFrame, *, available_capabilities: set[str] | None = None) -> list[Protocol]

Return every protocol that passes the three gates.

The shared filter for both select() (rule-based ranker) and any opt-in selector that wants to choose among already-qualified candidates. Gates:

frame.primary_goal in p.handles
p.risk_max >= frame.risk
set(p.requires_capabilities).issubset(available_capabilities)

Returns the survivors in registration order. Empty list is a valid result — callers raise :class:NoMatchingProtocolError when appropriate (this method is filter-only by design).

Source code in src/locus/router/protocol.py

def filter_candidates(
    self,
    frame: GoalFrame,
    *,
    available_capabilities: set[str] | None = None,
) -> list[Protocol]:
    """Return every protocol that passes the three gates.

    The shared filter for both ``select()`` (rule-based ranker) and
    any opt-in selector that wants to choose among already-qualified
    candidates. Gates:

    - ``frame.primary_goal in p.handles``
    - ``p.risk_max >= frame.risk``
    - ``set(p.requires_capabilities).issubset(available_capabilities)``

    Returns the survivors in registration order. Empty list is a
    valid result — callers raise :class:`NoMatchingProtocolError`
    when appropriate (this method is filter-only by design).
    """
    caps_available = available_capabilities or set()
    return [
        p
        for p in self._protocols.values()
        if frame.primary_goal in p.handles
        and p.risk_max >= frame.risk
        and set(p.requires_capabilities).issubset(caps_available)
    ]

select ¶

select(frame: GoalFrame, *, available_capabilities: set[str] | None = None) -> Protocol

Pick the best protocol for frame.

Filters by handles ∋ primary_goal ∧ risk_max ≥ frame.risk ∧ required_capabilities ⊆ available_capabilities, then ranks candidates by complexity-fit + cost.

Source code in src/locus/router/protocol.py

def select(
    self,
    frame: GoalFrame,
    *,
    available_capabilities: set[str] | None = None,
) -> Protocol:
    """Pick the best protocol for ``frame``.

    Filters by ``handles ∋ primary_goal`` ∧ ``risk_max ≥ frame.risk``
    ∧ ``required_capabilities ⊆ available_capabilities``, then ranks
    candidates by complexity-fit + cost.
    """
    if not self._protocols:
        raise NoMatchingProtocolError("No protocols registered.")

    caps_available = available_capabilities or set()
    candidates = self.filter_candidates(frame, available_capabilities=caps_available)
    if not candidates:
        raise NoMatchingProtocolError(
            f"No protocol handles primary_goal={frame.primary_goal!r} "
            f"at risk={frame.risk!r} with capabilities={sorted(caps_available)}.",
        )

    return min(candidates, key=lambda p: _rank_key(p, frame))

builtin_protocols ¶

builtin_protocols() -> list[Protocol]

The full eight-protocol catalogue. Compose your own :class:ProtocolRegistry from these (or a subset).

Source code in src/locus/router/protocol.py

def builtin_protocols() -> list[Protocol]:
    """The full eight-protocol catalogue. Compose your own
    :class:`ProtocolRegistry` from these (or a subset)."""
    return [
        Protocol(
            id="direct_response",
            description="Single-agent answer for direct questions and explanations.",
            handles=[TaskType.ANSWER, TaskType.EXPLAIN, TaskType.RESEARCH],
            primary_for=[TaskType.ANSWER, TaskType.EXPLAIN],
            risk_max=Risk.LOW,
            cost="low",
            latency="low",
            builder=_build_direct_response,
        ),
        Protocol(
            id="plan_execute_validate",
            description="Three-stage pipeline for tasks needing structure and validation.",
            handles=[
                TaskType.PLAN,
                TaskType.BUILD,
                TaskType.MODIFY,
                TaskType.GENERATE_CODE,
                TaskType.REMEDIATE,
            ],
            primary_for=[TaskType.PLAN, TaskType.BUILD, TaskType.MODIFY],
            risk_max=Risk.MEDIUM,
            cost="medium",
            latency="medium",
            builder=_build_plan_execute_validate,
        ),
        Protocol(
            id="specialist_fanout",
            description="Fan out to multiple specialists and correlate findings.",
            handles=[
                TaskType.DIAGNOSE,
                TaskType.COMPARE,
                TaskType.MONITOR,
                TaskType.COORDINATE,
                TaskType.RESEARCH,
            ],
            primary_for=[TaskType.DIAGNOSE, TaskType.MONITOR, TaskType.RESEARCH],
            risk_max=Risk.MEDIUM,
            cost="high",
            latency="high",
            builder=_build_specialist_fanout,
        ),
        Protocol(
            id="debate",
            description="Two debaters argue opposing positions; a judge picks the winner.",
            handles=[TaskType.COMPARE, TaskType.RESEARCH],
            primary_for=[TaskType.COMPARE],
            risk_max=Risk.LOW,
            cost="high",
            latency="high",
            builder=_build_debate,
        ),
        Protocol(
            id="codegen_test_validate",
            description=(
                "Loop a tool-using agent that must declare PASS/FAIL on every "
                "iteration; stops when output starts with PASS."
            ),
            handles=[TaskType.GENERATE_CODE, TaskType.BUILD],
            primary_for=[TaskType.GENERATE_CODE],
            risk_max=Risk.MEDIUM,
            cost="medium",
            latency="high",
            builder=_build_codegen_test_validate,
        ),
        Protocol(
            id="approval_gated_execution",
            description=(
                "Single-agent execution behind a forced approval interrupt — "
                "for risky REMEDIATE / MODIFY / ESCALATE actions."
            ),
            handles=[
                TaskType.REMEDIATE,
                TaskType.MODIFY,
                TaskType.ESCALATE,
            ],
            primary_for=[TaskType.ESCALATE, TaskType.REMEDIATE],
            risk_max=Risk.HIGH,
            cost="medium",
            latency="medium",
            builder=_build_approval_gated_execution,
        ),
        Protocol(
            id="a2a_delegate",
            description=(
                "Forward the task to a remote agent via the A2A protocol. "
                "Requires BuilderContext.a2a_endpoint to be set at compile time."
            ),
            handles=[TaskType.COORDINATE, TaskType.ESCALATE],
            primary_for=[],  # opt-in only
            risk_max=Risk.MEDIUM,
            cost="medium",
            latency="high",
            builder=_build_a2a_delegate,
        ),
        Protocol(
            id="handoff_chain",
            description=(
                "Sequential chain of one-tool agents — each link adds one fact "
                "and hands off to the next."
            ),
            handles=[TaskType.PLAN, TaskType.RESEARCH, TaskType.COORDINATE],
            primary_for=[TaskType.COORDINATE],
            risk_max=Risk.MEDIUM,
            cost="medium",
            latency="high",
            builder=_build_handoff_chain,
        ),
    ]

BuilderContext ¶

Bases: BaseModel

Bundle of inputs passed to every protocol builder.

Builders need the model, the tool-resolution view, and the skill catalogue; passing a typed context keeps the builder signature stable as we add follow-up protocols.

NoMatchingProtocolError ¶

Bases: LookupError

Raised by :meth:ProtocolRegistry.select when no protocol fits the frame.

Picker¶

The picker decides which protocol best matches a GoalFrame. LLMProtocolPicker is the default; PickedProtocol is the decision record returned to the compiler.

LLMProtocolPicker ¶

LLMProtocolPicker(*, model: Any, system_prompt: str | None = None)

Picks a protocol from a pre-filtered candidate set using an LLM.

Construct with a locus model (any provider) and an optional system_prompt override. The picker reuses the same Agent + output_schema machinery the router's extractor uses — one structured-output call per pick.

Parameters¶

model: Any locus model instance (an :class:~locus.Agent-compatible model object or model string). system_prompt: Optional override for the picker's system prompt. The default instructs the model on the rule hierarchy (canonical match, cost-fit, conservative fallback).

Example::

picker = LLMProtocolPicker(model=my_model)
compiler = CognitiveCompiler(
    protocols=registry,
    capabilities=capabilities,
    policy=PolicyGate(),
    model=my_model,
    protocol_picker=picker,
)

Source code in src/locus/router/picker.py

def __init__(
    self,
    *,
    model: Any,
    system_prompt: str | None = None,
) -> None:
    self.model = model
    self.system_prompt = system_prompt or _DEFAULT_SYSTEM_PROMPT

pick `async` ¶

pick(frame: GoalFrame, candidates: list[Protocol]) -> tuple[Protocol, str]

Pick one protocol from candidates for frame.

Returns a (protocol, rationale) tuple. The compiler threads the rationale into the router.protocol.selected event so the LLM's reasoning is observable.

Raises :class:PickerError when the model output is unparseable or names an id not in candidates — the compiler catches this and falls back to the rule-based ranker.

The caller is responsible for the candidate set being non-empty and already filtered; this method does not re-validate handles / risk / capabilities.

Source code in src/locus/router/picker.py

async def pick(
    self,
    frame: GoalFrame,
    candidates: list[Protocol],
) -> tuple[Protocol, str]:
    """Pick one protocol from ``candidates`` for ``frame``.

    Returns a ``(protocol, rationale)`` tuple. The compiler threads
    the rationale into the ``router.protocol.selected`` event so
    the LLM's reasoning is observable.

    Raises :class:`PickerError` when the model output is unparseable
    or names an id not in ``candidates`` — the compiler catches this
    and falls back to the rule-based ranker.

    The caller is responsible for the candidate set being non-empty
    and already filtered; this method does not re-validate handles
    / risk / capabilities.
    """
    # Import here so the picker module doesn't pull Agent into the
    # import graph of anyone who only uses ProtocolRegistry.
    from locus.agent.agent import Agent

    prompt = (
        "Goal frame:\n"
        f"{_format_frame(frame)}\n\n"
        "Candidate protocols (one of these ids must be picked):\n"
        f"{_format_candidates(candidates)}\n\n"
        "Pick exactly one protocol id and explain in one sentence."
    )

    picker_agent = Agent(
        model=self.model,
        system_prompt=self.system_prompt,
        output_schema=PickedProtocol,
    )
    result: Any = await asyncio.to_thread(picker_agent.invoke, prompt)
    parsed = result.parsed
    if not isinstance(parsed, PickedProtocol):
        raise PickerError(
            f"Picker did not return PickedProtocol. "
            f"parse_error={result.parse_error!r}, "
            f"message={result.message!r}"
        )

    by_id = {p.id: p for p in candidates}
    picked = by_id.get(parsed.protocol_id)
    if picked is None:
        raise PickerError(
            f"Picker returned unknown protocol_id={parsed.protocol_id!r}. "
            f"Candidates were {sorted(by_id.keys())}."
        )

    return picked, parsed.rationale

PickedProtocol ¶

Bases: BaseModel

Typed output schema for :class:LLMProtocolPicker.

Used as output_schema= on the picker agent so the model is forced to return a structured pick rather than free-form text.

PickerError ¶

Bases: RuntimeError

Raised by :class:LLMProtocolPicker when the model output is unusable (parse failure or returned id not in the candidate set).

The compiler catches this and falls back to the rule-based ranker after emitting a router.protocol.picker_fallback event. Users do not normally see this exception.

Policy gate¶

Pre-execution policy checks (cost, risk, capability requirements, optional human approval). PolicyVerdict either allows compilation to proceed or rejects with a structured reason.

PolicyGate ¶

PolicyGate(*, max_risk: Risk = Risk.HIGH, require_approval_above: Risk = Risk.MEDIUM)

Pre-flight risk check for a (frame, protocol) pair.

Two thresholds:

max_risk — anything strictly above is denied.
require_approval_above — anything strictly above (and within max_risk) is allowed but flagged for an approval interrupt.

Defaults: max_risk=Risk.HIGH (nothing denied by default) and require_approval_above=Risk.MEDIUM (HIGH-risk frames need explicit approval).

Source code in src/locus/router/policy.py

def __init__(
    self,
    *,
    max_risk: Risk = Risk.HIGH,
    require_approval_above: Risk = Risk.MEDIUM,
) -> None:
    self._max_risk = max_risk
    self._approval_above = require_approval_above

PolicyVerdict ¶

Bases: BaseModel

One of three outcomes from :meth:PolicyGate.check.

PolicyDeniedError ¶

Bases: RuntimeError

Raised when the policy gate refuses to compile a runnable.

Compiler¶

Turns a PickedProtocol + GoalFrame into an executable Runnable. Approval callbacks (HITL gates) plug in here.

CognitiveCompiler ¶

CognitiveCompiler(*, protocols: ProtocolRegistry, capabilities: CapabilityIndex, policy: PolicyGate, model: Any, skills: SkillIndex | None = None, a2a_endpoint: str | None = None, on_approval: ApprovalCallback | None = None, protocol_picker: LLMProtocolPicker | None = None)

Glue between protocols, capabilities, policy, and the model.

Parameters¶

protocols: :class:ProtocolRegistry populated with built-in or custom protocols. capabilities: :class:CapabilityIndex over the surrounding ToolRegistry. The index resolves capability ids to real tools at compile time. policy: :class:PolicyGate that runs between selection and build. model: A locus model instance (or model string) injected into every builder. Builders pass it to :class:~locus.Agent / specialist constructors. skills: Optional :class:SkillIndex. When provided, every emitted :class:Agent is configured with a :class:~locus.skills.SkillsPlugin containing the skills tagged for frame.domain (plus any globally-tagged skills). The agent loop's L1 / L2 / L3 progressive disclosure surfaces them at runtime. on_approval: Optional async callback fired when the verdict requires approval. Defaults to denying — wire your workbench / CLI approval flow here. protocol_picker: Optional :class:LLMProtocolPicker. When present, the compiler delegates the last-mile protocol pick to the model whenever the filter leaves more than one candidate. When absent (the default), selection is the deterministic :func:_rank_key-based ranker. The picker is the only part of the routing pipeline that uses the model; everything else — filtering, policy gating, capability binding, builder dispatch — remains rule-based.

Source code in src/locus/router/compiler.py

def __init__(
    self,
    *,
    protocols: ProtocolRegistry,
    capabilities: CapabilityIndex,
    policy: PolicyGate,
    model: Any,
    skills: SkillIndex | None = None,
    a2a_endpoint: str | None = None,
    on_approval: ApprovalCallback | None = None,
    protocol_picker: LLMProtocolPicker | None = None,
) -> None:
    self.protocols = protocols
    self.capabilities = capabilities
    self.policy = policy
    self.model = model
    self.skills = skills
    self.a2a_endpoint = a2a_endpoint
    self._on_approval: ApprovalCallback = on_approval or _default_deny
    self.protocol_picker = protocol_picker

compile `async` ¶

compile(frame: GoalFrame, run_id: str | None = None) -> Runnable

Pick a protocol, run the gate, build the runnable.

run_id (when provided) scopes every emitted :class:StreamEvent so the workbench's SSE consumer can correlate selection / verdict / compile events with one cognitive dispatch.

Source code in src/locus/router/compiler.py

async def compile(self, frame: GoalFrame, run_id: str | None = None) -> Runnable:
    """Pick a protocol, run the gate, build the runnable.

    ``run_id`` (when provided) scopes every emitted
    :class:`StreamEvent` so the workbench's SSE consumer can
    correlate selection / verdict / compile events with one
    cognitive dispatch.
    """
    available = {c.id for c in self.capabilities.all()}
    candidates = self.protocols.filter_candidates(frame, available_capabilities=available)
    if not candidates:
        err = NoMatchingProtocolError(
            f"No protocol handles primary_goal={frame.primary_goal!r} "
            f"at risk={frame.risk!r} with capabilities={sorted(available)}.",
        )
        if run_id:
            await emit_protocol_no_match(run_id, frame, str(err))
        raise err

    protocol, method, rationale = await self._pick_protocol(frame, candidates, run_id=run_id)

    if run_id:
        await emit_protocol_selected(
            run_id, frame, protocol, method=method, rationale=rationale
        )

    verdict = self.policy.check(frame, protocol)
    if run_id:
        await emit_policy_verdict(run_id, frame, protocol, verdict)
    if not verdict.allow:
        raise PolicyDeniedError(verdict.reason)

    # Intersect with the actual registry — the LLM extractor sometimes
    # hallucinates capability ids that don't exist. Strict lookup
    # would crash the whole dispatch on a single bad id; lenient
    # filtering is the right call at this boundary because the
    # protocol registry's selection step has *already* checked that
    # every protocol-required capability is available, so the only
    # things being dropped here are extractor-suggested extras.
    requested = list(frame.required_capabilities)
    valid = [cid for cid in requested if cid in available]
    caps = self.capabilities.lookup(valid) if valid else []
    runnable = protocol.builder(frame, caps, self._build_context())

    if verdict.require_approval:
        runnable = _ApprovalRunnable(
            inner=runnable,
            frame=frame,
            verdict=verdict,
            callback=self._on_approval,
        )

    if run_id:
        await emit_runnable_compiled(
            run_id,
            protocol_id=protocol.id,
            runnable_type=type(runnable).__name__,
            capability_count=len(caps),
        )
    return runnable

ApprovalCallback `module-attribute` ¶

ApprovalCallback = Callable[[GoalFrame, PolicyVerdict], Awaitable[bool]]

Async callback used when a verdict requires approval.

Returning True lets the compiled runnable execute; returning False raises :class:PolicyDeniedError. Defaults deny.

Runnables¶

The compiled execution shape. Runnable is the base Protocol — every concrete runnable below produces a RunnableResult when invoked.

Runnable ¶

Bases: Protocol

Anything the router compiler can return.

The contract is intentionally minimal: a single async execute that yields a normalized :class:RunnableResult. Builders wrap each emitted primitive in the matching adapter below — call sites never branch on the concrete primitive type.

AgentRunnable ¶

Bases: BaseModel

Adapter from :class:locus.Agent to :class:Runnable.

Agent.invoke is sync but already handles being called from a running event loop (it spawns a worker thread internally). We wrap it in :func:asyncio.to_thread so the surrounding coroutine yields properly while the agent runs.

PipelineRunnable ¶

Bases: BaseModel

Adapter from any composition primitive to :class:Runnable.

Works for :class:SequentialPipeline, :class:ParallelPipeline, and :class:LoopAgent — they all share the async run(task) -> PipelineResult shape.

OrchestratorRunnable ¶

Bases: BaseModel

Adapter from :class:locus.Orchestrator to :class:Runnable.

DebateRunnable ¶

Bases: BaseModel

Adapter for the debate protocol.

Runs N debater agents (each a :class:Agent) in parallel against the same task, then feeds the joined transcript to a single judge :class:Agent that picks the strongest argument. The judge's verdict is the surface text. There is no native locus "debate" primitive — this builder composes :class:ParallelPipeline + an extra :class:Agent step, which is small enough to live in this adapter.

A2ARunnable ¶

Bases: BaseModel

Adapter for the a2a_delegate protocol.

Wraps an :class:A2AClient so a remote, separately-deployed agent can serve a request. The remote agent is responsible for any tool calls / orchestration on its end; the router simply forwards the user prompt and packages the response.

Capabilities¶

The skill / tool index the compiler binds protocol slots to. Custom capabilities register themselves via CapabilityIndex.annotate(...) or by exposing a SkillIndex.

Capability ¶

Bases: BaseModel

A tool (or human action) tagged with router-relevant metadata.

is_human `property` ¶

is_human: bool

True if this capability is the human-approval sentinel.

CapabilityIndex ¶

CapabilityIndex(tools: ToolRegistry)

View over a :class:ToolRegistry that adds router metadata.

Not a replacement for ToolRegistry — the underlying Tool instances live there. CapabilityIndex only holds the metadata overlay (domain, risk) and resolves capabilities back to tools on demand.