codenib.mcp

CodeNib MCP server - exposes backbone capabilities over stdio.

Provides MCP tools for semantic indexing, CodeGraph, and hybrid retrieval (vector, BM25, regex, Zoekt trigram) for external agent frameworks.

Modules:

Name	Description
context	ServerContext - loads a RepoManifest and selected index objects from disk.
explore_bounds	Deterministic byte bounds for composed MCP exploration responses.
explore_session	Bounded, fingerprint-safe context deduplication for stdio MCP sessions.
prompts	MCP prompt resource - guidance for calling agents.
server	CodeNib MCP server - stdio transport.
tool_surface	Configurable MCP tool visibility for agent-surface experiments.
tools	MCP tool implementations.

Classes:

Name	Description
ServerContext	Runtime context for the MCP server.

ServerContext dataclass

ServerContext(
    manifest: RepoManifest,
    symbol_graph: CodeGraph | None = None,
    bm25: BM25CodeIndexer | None = None,
    regex_index: RegexNodeIndex | None = None,
    zoekt: ZoektSearcher | None = None,
    vector: CodeVectorStore | None = None,
    lsp_provider: Any | None = None,
    lsp_provider_selection: dict[str, Any] = dict(),
    errors: dict[str, str] = dict(),
    artifact: Mapping[str, Any] | None = None,
    source_error: str | None = "source binding has not been verified",
    _native_index_authorization: NativeIndexAuthorization | None = None,
    _artifact_binding: ContextArtifactBinding | None = None,
    _source_binding: RepositorySourceBinding | None = None,
)

Runtime context for the MCP server.

Holds the loaded manifest and runtime-loaded index objects. Missing or failed indexes stay None; tools check at call time and return descriptive errors.

Methods:

Name	Description
read_source_bytes	Read one repository file through the retained source authority.
borrow_source_reader	Borrow the exact source reader retained by this context owner.
verify_source_status	Refresh whole-tree source truth before publishing verified status.
load	Load a manifest and the selected runtime views.
load_views	Load additional manifest views without disturbing loaded resources.
close	Release runtime resources owned by this context.
begin_explore_session	Create fresh state for one stdio connection.
ensure_explore_session	Return the connection runtime, creating one for direct embeddings.
end_explore_session	Discard one connection's ledger without clearing a newer runtime.
configure_lsp_provider	Bind a runtime-only provider without mutating persisted artifacts.
validate_views	Open selected artifacts without initializing a vector query model.

Attributes:

Name	Type	Description
source_verified	bool	Return whether live reads retain exact content-byte authority.
source_verification_scope	str | None	M1 authenticates v2 content bytes, never mutable Git HEAD state.
commit_verified	bool	Mutable checkout commit provenance requires an M2 source snapshot.
loaded_views	frozenset[str]	Return the runtime views currently available in this context.

source_verified property

source_verified: bool

Return whether live reads retain exact content-byte authority.

source_verification_scope property

source_verification_scope: str | None

M1 authenticates v2 content bytes, never mutable Git HEAD state.

commit_verified property

commit_verified: bool

Mutable checkout commit provenance requires an M2 source snapshot.

loaded_views property

loaded_views: frozenset[str]

Return the runtime views currently available in this context.

read_source_bytes

read_source_bytes(relative: str, *, max_bytes: int) -> bytes

Read one repository file through the retained source authority.

Source code in codenib/mcp/context.py

def read_source_bytes(self, relative: str, *, max_bytes: int) -> bytes:
    """Read one repository file through the retained source authority."""

    if self._source_binding is None:
        raise RuntimeError(
            f"source reads are unavailable: {self.source_error or 'unverified'}"
        )
    try:
        return self._source_binding.read_bytes(relative, max_bytes=max_bytes)
    except Exception as exc:
        self.source_error = str(exc)
        raise

borrow_source_reader

borrow_source_reader() -> RepositorySourceReader

Borrow the exact source reader retained by this context owner.

Source code in codenib/mcp/context.py

def borrow_source_reader(self) -> RepositorySourceReader:
    """Borrow the exact source reader retained by this context owner."""

    if self._source_binding is None or not self._source_binding.usable:
        raise RuntimeError(
            f"source reads are unavailable: {self.source_error or 'unverified'}"
        )
    return self._source_binding.borrow_reader()

verify_source_status

verify_source_status() -> bool

Refresh whole-tree source truth before publishing verified status.

Source code in codenib/mcp/context.py

def verify_source_status(self) -> bool:
    """Refresh whole-tree source truth before publishing verified status."""

    if self._source_binding is None:
        return False
    try:
        self._source_binding.authenticated_identity_snapshot()
    except Exception as exc:
        self.source_error = self._source_binding.failure_reason or str(exc)
        return False
    self.source_error = None
    return True

load classmethod

load(
    manifest_path: RepoManifest | str | Path,
    *,
    views: Iterable[str] | None = None,
    artifact: Mapping[str, Any] | None = None,
    artifact_binding: ContextArtifactBinding | None = None,
    artifact_reader: PublicationDirectoryReader | None = None,
    native_index_authorization: NativeIndexAuthorization | None = None,
    source_binding: RepositorySourceBinding | None = None,
    _context_owner: Callable[[ServerContext], None] | None = None
) -> ServerContext

Load a manifest and the selected runtime views.

views=None preserves the MCP server's load-all behavior. Explicit selections avoid importing or starting unrelated view runtimes. Each selected view is loaded independently; a failure in one does not block the others. Failed views are recorded in errors.

Source code in codenib/mcp/context.py

@classmethod
def load(
    cls,
    manifest_path: RepoManifest | str | Path,
    *,
    views: Iterable[str] | None = None,
    artifact: Mapping[str, Any] | None = None,
    artifact_binding: ContextArtifactBinding | None = None,
    artifact_reader: PublicationDirectoryReader | None = None,
    native_index_authorization: NativeIndexAuthorization | None = None,
    source_binding: RepositorySourceBinding | None = None,
    _context_owner: Callable[[ServerContext], None] | None = None,
) -> ServerContext:
    """Load a manifest and the selected runtime views.

    ``views=None`` preserves the MCP server's load-all behavior. Explicit
    selections avoid importing or starting unrelated view runtimes. Each
    selected view is loaded independently; a failure in one does not block
    the others. Failed views are recorded in ``errors``.
    """
    ctx: ServerContext | None = None
    try:
        if (
            artifact_reader is not None
            and type(artifact_reader) is not PublicationDirectoryReader
        ):
            raise TypeError(
                "authenticated artifact reader must use the exact reader type"
            )
        if artifact_reader is not None and artifact_binding is None:
            raise ValueError(
                "an authenticated artifact reader requires its verified binding"
            )
        selected = _resolve_views(views)
        artifact_origin = artifact is not None or artifact_binding is not None
        if artifact_origin and native_index_authorization is not None:
            raise ValueError(
                "portable artifact contexts cannot authorize native vector parsing"
            )
        if artifact_origin:
            disallowed = selected - _PORTABLE_ARTIFACT_RUNTIME_VIEWS
            if views is not None and disallowed:
                raise ValueError(
                    "portable artifact contexts cannot load native views: "
                    + ", ".join(sorted(disallowed))
                )
            selected &= _PORTABLE_ARTIFACT_RUNTIME_VIEWS
        if artifact_binding is not None:
            if (
                not isinstance(manifest_path, RepoManifest)
                or manifest_path.to_dict() != artifact_binding.manifest.to_dict()
            ):
                raise ValueError(
                    "artifact runtime manifest must come from its verified binding"
                )
            manifest = artifact_binding.manifest
        else:
            manifest = (
                manifest_path
                if isinstance(manifest_path, RepoManifest)
                else RepoManifest.load(manifest_path)
            )

        ctx = cls(
            manifest=manifest,
            artifact=dict(artifact) if artifact is not None else None,
            _native_index_authorization=native_index_authorization,
            _artifact_binding=artifact_binding,
            _source_binding=None,
        )
        if _context_owner is not None:
            if not callable(_context_owner):
                raise TypeError("context owner must be callable")
            # Publish the context in a stable caller-owned slot before it
            # acquires any runtime resource.  A cancellation after a view
            # loader returns can then be reconciled by that owner instead
            # of dropping an unreturned live context.
            _context_owner(ctx)
        if artifact_binding is not None:
            artifact_binding.install_source_binding(
                ctx._install_repository_source,
                expected=source_binding,
                require_expected=source_binding is not None,
            )
        else:
            ctx._install_repository_source(source_binding)
        owned_source = ctx._source_binding

        if owned_source is not None:
            from ..compiler.manifest_source import require_manifest_source_identity
            from ..source_fingerprint import (
                is_secure_source_fingerprint_v2,
                lexical_repository_path,
            )

            source_identity = owned_source.authenticated_identity_snapshot()
            if (
                not owned_source.usable
                or not is_secure_source_fingerprint_v2(manifest.source_fingerprint)
                or source_identity.root
                != lexical_repository_path(manifest.repo_path)
            ):
                raise ValueError(
                    "repository source authority does not match the manifest"
                )
            require_manifest_source_identity(
                source_identity,
                manifest,
                label="repository source authority",
                mismatch_message=(
                    "repository source authority does not match the manifest"
                ),
            )

        ctx.source_error = (
            None
            if owned_source is not None
            else "source binding has not been verified"
        )

        if artifact_reader is None:
            ctx.load_views(selected)
        else:
            ctx.load_views(selected, artifact_reader=artifact_reader)
        ctx.configure_lsp_provider(
            allow_native=False,
            native_disabled_reason=(
                "portable_artifact_uses_persisted_graph"
                if artifact_origin
                else "local_source_not_verified"
            ),
        )

        cap_summary = {k: v for k, v in manifest.capabilities.items() if v}
        loaded = [
            view
            for view, _ in _VIEW_LOADERS
            if getattr(ctx, view, None) is not None
        ]
        logger.info(
            "ServerContext ready  repo=%s  commit=%s  requested=%s  loaded=%s  "
            "capabilities=%s  errors=%s",
            manifest.repo_path,
            manifest.commit[:8] if manifest.commit else "N/A",
            sorted(selected),
            loaded or "none",
            cap_summary or "none",
            list(ctx.errors) or "none",
        )
        return ctx
    except BaseException as primary:  # noqa: B036 - preserve primary
        from ..artifacts.runtime import (
            _raise_source_cleanup_failure,
            _source_cleanup_owner_is_pending,
        )

        runtime_cleanup_failure: BaseException | None = None
        if ctx is not None:
            try:
                ctx.close()
            except BaseException as exc:  # noqa: B036 - retain source priority
                _attach_context_cleanup_owner(exc, ctx)
                _attach_context_cleanup_owner(primary, ctx)
                runtime_cleanup_failure = exc
        cleanup_source = (
            ctx._source_binding
            if ctx is not None and ctx._source_binding is not None
            else (
                source_binding
                if artifact_binding is None
                and _source_cleanup_owner_is_pending(source_binding)
                else None
            )
        )
        cleanup_failure = runtime_cleanup_failure
        if cleanup_source is not None:
            try:
                cleanup_source.close()
            except BaseException as exc:  # noqa: B036 - apply shared priority
                cleanup_failure = _retain_context_cleanup_failure(
                    cleanup_failure,
                    "source cleanup retry also failed",
                    exc,
                )
        pending_owner = (
            cleanup_source
            if _source_cleanup_owner_is_pending(cleanup_source)
            else None
        )
        _raise_source_cleanup_failure(
            primary,
            cleanup_failure,
            pending_owner,
        )

load_views

load_views(
    views: Iterable[str],
    *,
    native_index_authorization: NativeIndexAuthorization | None = None,
    artifact_reader: PublicationDirectoryReader | None = None
) -> dict[str, str]

Load additional manifest views without disturbing loaded resources.

View dependencies are resolved in the same way as :meth:load. The operation is idempotent and serialized so query-time planners can safely request only the backends selected for a query. The returned mapping contains requested views that remain unavailable.

Source code in codenib/mcp/context.py

def load_views(
    self,
    views: Iterable[str],
    *,
    native_index_authorization: NativeIndexAuthorization | None = None,
    artifact_reader: PublicationDirectoryReader | None = None,
) -> Dict[str, str]:
    """Load additional manifest views without disturbing loaded resources.

    View dependencies are resolved in the same way as :meth:`load`. The
    operation is idempotent and serialized so query-time planners can
    safely request only the backends selected for a query. The returned
    mapping contains requested views that remain unavailable.
    """

    selected = _resolve_views(views)
    with self._view_lock:
        if (
            artifact_reader is not None
            and type(artifact_reader) is not PublicationDirectoryReader
        ):
            raise TypeError(
                "authenticated artifact reader must use the exact reader type"
            )
        if artifact_reader is not None and self._artifact_binding is None:
            raise ValueError(
                "an authenticated artifact reader requires its verified binding"
            )
        artifact_origin = (
            self.artifact is not None or self._artifact_binding is not None
        )
        if artifact_origin:
            disallowed = selected - _PORTABLE_ARTIFACT_RUNTIME_VIEWS
            if disallowed:
                raise ValueError(
                    "portable artifact contexts cannot load native views: "
                    + ", ".join(sorted(disallowed))
                )
            if (
                native_index_authorization is not None
                or self._native_index_authorization is not None
            ):
                raise ValueError(
                    "portable artifact contexts cannot authorize native parsing"
                )
        if native_index_authorization is not None:
            self._native_index_authorization = native_index_authorization
        for view, loader_name in _VIEW_LOADERS:
            if view not in selected or getattr(self, view, None) is not None:
                continue
            if view == "bm25" and artifact_reader is not None:
                self._load_bm25(artifact_reader=artifact_reader)
            else:
                getattr(self, loader_name)()
            if getattr(self, view, None) is not None:
                self.errors.pop(view, None)
        if "symbol_graph" in selected:
            self.configure_lsp_provider(
                allow_native=self._lsp_allow_native,
                native_disabled_reason=self._lsp_native_disabled_reason,
            )
        return {
            view: self.errors.get(view, "view did not load")
            for view in selected
            if getattr(self, view, None) is None
        }

close

close() -> None

Release runtime resources owned by this context.

Source code in codenib/mcp/context.py

def close(self) -> None:
    """Release runtime resources owned by this context."""

    with self._view_lock:
        deferred: BaseException | None = None
        try:
            self.end_explore_session()
        except BaseException as exc:  # noqa: B036 - visit every runtime owner
            deferred = exc
        self.lsp_provider = None
        if self.zoekt is not None:
            try:
                _stop_zoekt(self.zoekt)
            except BaseException as exc:  # noqa: B036 - continue cleanup
                deferred = _retain_context_cleanup_failure(
                    deferred,
                    "Zoekt cleanup also failed",
                    exc,
                )
            else:
                self.zoekt = None
        if self.zoekt is None and self._zoekt_snapshot is not None:
            try:
                self._zoekt_snapshot.close()
            except BaseException as exc:  # noqa: B036 - continue cleanup
                deferred = _retain_context_cleanup_failure(
                    deferred,
                    "Zoekt shard snapshot cleanup also failed",
                    exc,
                )
            if self._zoekt_snapshot.closed:
                self._zoekt_snapshot = None
        if self.vector is not None:
            try:
                _close_vector(self.vector)
            except BaseException as exc:  # noqa: B036 - continue cleanup
                deferred = _retain_context_cleanup_failure(
                    deferred,
                    "vector cleanup also failed",
                    exc,
                )
            else:
                self.vector = None
        if self._source_binding is not None:
            try:
                self._source_binding.close()
            except BaseException as exc:  # noqa: B036 - preserve retry owner
                deferred = _retain_context_cleanup_failure(
                    deferred,
                    "source cleanup also failed",
                    exc,
                )
            if self._source_binding.closed:
                self._source_binding = None
                self.source_error = "source binding is closed"
        if deferred is not None:
            _attach_context_cleanup_owner(deferred, self)
            raise deferred

begin_explore_session

begin_explore_session() -> ExploreSessionRuntime

Create fresh state for one stdio connection.

Stdio normally has one live connection per process. Replacing any previous runtime keeps reconnects fail-closed even when an embedding application reuses the same :class:ServerContext.

Source code in codenib/mcp/context.py

def begin_explore_session(self) -> ExploreSessionRuntime:
    """Create fresh state for one stdio connection.

    Stdio normally has one live connection per process. Replacing any
    previous runtime keeps reconnects fail-closed even when an embedding
    application reuses the same :class:`ServerContext`.
    """

    from .explore_session import ExploreSessionRuntime

    with self._view_lock:
        previous = self.explore_runtime
        runtime = ExploreSessionRuntime()
        self.explore_runtime = runtime
        self._explore_loop = _running_event_loop()
        if previous is not None:
            previous.close()
        return runtime

ensure_explore_session

ensure_explore_session() -> ExploreSessionRuntime

Return the connection runtime, creating one for direct embeddings.

Source code in codenib/mcp/context.py

def ensure_explore_session(self) -> ExploreSessionRuntime:
    """Return the connection runtime, creating one for direct embeddings."""

    from .explore_session import ExploreSessionRuntime

    with self._view_lock:
        loop = _running_event_loop()
        if self.explore_runtime is None or (
            loop is not None
            and self._explore_loop is not None
            and loop is not self._explore_loop
        ):
            previous = self.explore_runtime
            self.explore_runtime = ExploreSessionRuntime()
            if previous is not None:
                previous.close()
        if loop is not None:
            self._explore_loop = loop
        return self.explore_runtime

end_explore_session

end_explore_session(runtime: ExploreSessionRuntime | None = None) -> None

Discard one connection's ledger without clearing a newer runtime.

Source code in codenib/mcp/context.py

def end_explore_session(self, runtime: ExploreSessionRuntime | None = None) -> None:
    """Discard one connection's ledger without clearing a newer runtime."""

    with self._view_lock:
        current = self.explore_runtime
        if current is None or (runtime is not None and current is not runtime):
            return
        current.close()
        self.explore_runtime = None
        self._explore_loop = None

configure_lsp_provider

configure_lsp_provider(
    *,
    allow_native: bool,
    native_disabled_reason: str = "native_provider_not_authorized"
) -> dict[str, Any]

Bind a runtime-only provider without mutating persisted artifacts.

Source code in codenib/mcp/context.py

def configure_lsp_provider(
    self,
    *,
    allow_native: bool,
    native_disabled_reason: str = "native_provider_not_authorized",
) -> Dict[str, Any]:
    """Bind a runtime-only provider without mutating persisted artifacts."""

    from ..agent.lsp_provider import select_checkout_lsp_provider

    self._lsp_allow_native = allow_native
    self._lsp_native_disabled_reason = native_disabled_reason
    provider, selection = select_checkout_lsp_provider(
        project_root=self.manifest.repo_path,
        languages=self.manifest.languages,
        symbol_graph=self.symbol_graph,
        source_selection=(
            self.manifest.source_selection or DEFAULT_REPOSITORY_SOURCE_SELECTION
        ),
        allow_native=allow_native,
        native_disabled_reason=native_disabled_reason,
    )
    self.lsp_provider = provider
    self.lsp_provider_selection = selection
    return dict(selection)

validate_views classmethod

validate_views(
    manifest: RepoManifest | str | Path,
    *,
    views: Iterable[str],
    native_index_authorization: NativeIndexAuthorization | None = None
) -> dict[str, str]

Open selected artifacts without initializing a vector query model.

The returned mapping contains only unavailable views. Vector indexes follow the normal FAISS/document load path with a fixed-dimension embedding probe. Temporary vector and Zoekt resources are released before returning.

Source code in codenib/mcp/context.py

@classmethod
def validate_views(
    cls,
    manifest: RepoManifest | str | Path,
    *,
    views: Iterable[str],
    native_index_authorization: NativeIndexAuthorization | None = None,
) -> Dict[str, str]:
    """Open selected artifacts without initializing a vector query model.

    The returned mapping contains only unavailable views. Vector indexes
    follow the normal FAISS/document load path with a fixed-dimension
    embedding probe. Temporary vector and Zoekt resources are released
    before returning.
    """

    selected = _resolve_views(views)
    resolved_manifest = (
        manifest
        if isinstance(manifest, RepoManifest)
        else RepoManifest.load(manifest)
    )
    ctx = cls(
        manifest=resolved_manifest,
        _native_index_authorization=native_index_authorization,
    )
    available: set[str] = set()
    try:
        for view, loader_name in _VIEW_LOADERS:
            if view not in selected:
                continue
            loader = getattr(ctx, loader_name)
            if view == "vector":
                loader(probe=True)
            else:
                loader()
            if getattr(ctx, view, None) is not None:
                available.add(view)
        return {
            view: ctx.errors.get(view, "view did not load")
            for view in selected
            if view not in available
        }
    finally:
        active_failure = sys.exc_info()[1]
        try:
            ctx.close()
        except BaseException as cleanup_failure:  # noqa: B036 - keep priority
            if active_failure is None:
                raise
            preferred = _retain_context_cleanup_failure(
                active_failure,
                "view validation cleanup also failed",
                cleanup_failure,
            )
            _attach_context_cleanup_owner(preferred, ctx)
            if preferred is cleanup_failure:
                raise cleanup_failure from active_failure