"""Expose OKF bundles over the Model Context Protocol, in two shapes. Beside `skill.py` because it belongs to the same class: a way to put a bundle in front of an agent. The skill hands a consumer a document telling it which command to run; this hands it a set of tools a client calls. Neither ranks anything of its own -- both reach `consume.build_payload`, which stays the one reading direction this library has. TWO SHAPES, ONE IMPLEMENTATION. * `--bundle PATH` serves exactly ONE bundle, fixed at startup. The bundle tools take no bundle argument, because there is nothing to choose. * `--root PATH` (repeatable) serves every bundle found under the roots, and knows NONE of them by name. Discovery happens per call, so a bundle added, removed or rebuilt while the process runs is seen by the next call without a restart, a configuration edit or a code change. NOTHING IS CACHED ACROSS CALLS, AND THAT IS THE DESIGN RATHER THAN AN OVERSIGHT. A server that read the bundle list once at startup would keep answering after the bundle was rebuilt, with an identity that no longer describes the bytes -- and an answer from yesterday's bundle is the one failure a consumer cannot see from the outside. Every call re-walks the roots and recomputes `bundle_ref`, so the identity in an answer is always a fact about the bytes on disk at the moment of the call. The cost is real: the identity is a sha256 over the whole concept tree, and it is paid per call. WHY THE PROTOCOL IS WRITTEN HERE AND NOT TAKEN FROM AN SDK. This package declares exactly one runtime dependency, the security guard, and `tests/test_packaging.py::test_the_only_runtime_dependency_is_the_security_boundary` pins that list literally. An MCP SDK would be the second, on the DEFAULT install path, for four JSON-RPC methods and a newline framing -- so the protocol is written narrowly, with stdlib only, and the packaging invariant stays a fact rather than an intention. Chosen over the SDK because the surface needed is `initialize`, `notifications/initialized`, `tools/list` and `tools/call`, and nothing here needs resources, prompts, sampling or progress. CONTAINMENT IS TWO INDEPENDENT CHECKS, NEVER ONE. A concept is reachable only if the bundle's own index names it (`consume.enumerate_concepts`, which refuses a target climbing above the root) AND its resolved path is inside the bundle (`connectors.safe_resolve`, on canonical paths). Either alone would be defensible; the pair is what makes a defect in one of them survivable. AND IT IS EVERY READ PATH, not the one tool that happened to have it. Until `consume.resolve_in_bundle` existed, the second check was made by `okf_fetch` alone: `okf_ask` and `okf_describe` joined the index's own name onto the root and opened whatever was there, so a link out of the bundle was read and delivered. The index rule is a STRING rule -- it cannot see a symlink -- which is exactly why one of the two checks is not enough. """ from __future__ import annotations import argparse import json import sys from collections.abc import Iterator, Mapping, Sequence from dataclasses import dataclass from pathlib import Path from typing import Any, TextIO from . import consume as okf_consume from . import materialize from .errors import SourceError from .profiles import BundleProfile #: The revision this server implements. A client asking for another is #: answered with this one, which the specification permits: the client then #: decides whether it can proceed. PROTOCOL_VERSION = "2025-06-18" SERVER_NAME = "okf" #: What a client keeps of `instructions` and of each tool description. Claude #: Code truncates BOTH at 2 KB (`docs/en/mcp`), and truncation is worse than #: rejection here: a reader gets the first half of a method and no sign that #: the rest existed. The long form of the working method lives in the skill, #: which has no such cap. CLIENT_TRUNCATION_BYTES = 2048 #: The SHORT working method, and the reason it is here rather than only in the #: skill: **a subagent inherits its session's MCP tools and not its skills.** #: So a method stated only in a skill reaches the main thread and no arm below #: it, and these few hundred bytes are the one place every caller sees. Held #: under the cap by a test, with a control so the assertion is a measurement. SERVER_INSTRUCTIONS = ( "Bundles are read-only and no call here runs a model.\n\n" "HOW TO USE THIS SERVER. Read the bundle's map first with `okf_describe`, " "then put the question into the bundle's own words -- its documents may be " "written in another language than the question, and the ranking matches " "words. Split a broad question into two to four sub-questions and call " "`okf_ask` once per sub-question. After each call read BOTH what came back " "and what lay just outside the cut: `withheld.nearest` names the " "best-ranked concepts that missed, with their titles. If one of them is " "what you wanted, that is a fact about the WORDS, not a closed door -- ask " "again with that concept's own words, or fetch it by name with " "`okf_fetch`. Several calls are normal and expected; there is no limit and " "no penalty. Then write ONE answer, ordered by sub-question, in the " "questioner's language and in ordinary prose, citing the document and the " "section (and the bundle, when you read more than one). Say plainly what " "the bundles do not cover.\n\n" "Every excerpt carries the bundle id and concept id a claim must be " "attributed to; the payload states what it withheld and why." ) #: How deep a root is walked looking for bundles. A bundle is a directory with #: an `index.md` carrying a `bundle_id`, and the walk does NOT descend into one #: it has found -- a bundle inside a bundle is the door's own collision case, #: not a second bundle. Bounded rather than unbounded because a root is given #: by an operator and may be a home directory by accident. MAX_DISCOVERY_DEPTH = 3 #: The largest concept `okf_fetch` will hand over whole. A concept is a #: section of a document; this is two orders of magnitude above the largest in #: any bundle measured here, and it exists so that a bundle carrying a file #: that is not a concept cannot turn one tool call into a memory cost the #: caller never asked for. Refused with its own code, never truncated: a #: truncated concept read as whole is a wrong answer that looks right. MAX_CONCEPT_BYTES = 1024 * 1024 #: Default breadth of an `okf_ask`. The library's own default, restated here #: rather than imported implicitly, because a tool's default is part of its #: contract. DEFAULT_K = okf_consume.DEFAULT_K class ToolError(Exception): """A refusal a client can act on. Always loud: it leaves the server as a JSON-RPC error, never as a plausible-looking empty answer.""" def __init__(self, message: str, *, code: str) -> None: super().__init__(message) self.code = code # -------------------------------------------------------------------------- # Discovery # -------------------------------------------------------------------------- @dataclass(frozen=True) class Served: bundle_id: str root: Path @dataclass(frozen=True) class Unreadable: """A directory that looks like a bundle and cannot be read as one. Reported rather than skipped. A broken manifest that simply vanishes from the list is an absence with no denominator, and the caller cannot tell it from a bundle that was never there. """ path: str reason: str @dataclass(frozen=True) class Discovery: bundles: tuple[Served, ...] unreadable: tuple[Unreadable, ...] def _declared_bundle_id(index: Path) -> str: frontmatter = materialize.parse_frontmatter(index) return str(frontmatter.get("bundle_id", "")).strip() def _walk(root: Path, depth: int) -> Iterator[Path]: """Directories under `root`, breadth-first, to `MAX_DISCOVERY_DEPTH`. A symlink is never descended and never yielded: a link inside a served root pointing outside it is exactly how a root boundary is escaped, and refusing to follow one is cheaper than proving each target is contained. """ if depth > MAX_DISCOVERY_DEPTH: return try: entries = sorted(root.iterdir(), key=lambda path: path.name) except OSError: return for entry in entries: if entry.is_symlink() or not entry.is_dir(): continue yield entry if not (entry / "index.md").is_file(): yield from _walk(entry, depth + 1) def _candidates(roots: Sequence[Path], *, include_roots: bool) -> Iterator[Path]: """Directories to test for being a bundle. `include_roots` is the whole difference between the two shapes at this level: `--bundle` points AT a bundle, `--root` points at a directory that holds them. Without it the one-to-one server discovers its own children and never itself -- which is how the first build of this module answered every call with "the bundle this server was started on is no longer readable". """ for root in roots: if include_roots: yield root else: yield from _walk(root, 1) def discover(roots: Sequence[Path], *, include_roots: bool = False) -> Discovery: """Every bundle under the roots, recomputed on every call.""" bundles: dict[str, Served] = {} unreadable: list[Unreadable] = [] for candidate in _candidates(roots, include_roots=include_roots): index = candidate / "index.md" if not index.is_file(): continue try: bundle_id = _declared_bundle_id(index) except (OSError, UnicodeDecodeError, ValueError) as error: unreadable.append(Unreadable(candidate.name, f"index.md unreadable: {error}")) continue if not bundle_id: unreadable.append(Unreadable(candidate.name, "index.md declares no bundle_id")) continue if bundle_id in bundles: unreadable.append( Unreadable(candidate.name, f"a second bundle claims the id `{bundle_id}`") ) continue bundles[bundle_id] = Served(bundle_id, candidate) return Discovery( tuple(bundles[key] for key in sorted(bundles)), tuple(sorted(unreadable, key=lambda entry: entry.path)), ) @dataclass(frozen=True) class Surface: """What the two shapes have in common, with the difference in one flag.""" roots: tuple[Path, ...] fixed: str | None profile: BundleProfile @property def one_to_many(self) -> bool: return self.fixed is None def discovery(self) -> Discovery: """Re-read on every call, in both shapes. The one-to-one server tests its own root; the one-to-many server tests what is under its roots.""" return discover(self.roots, include_roots=not self.one_to_many) def resolve(self, bundle_id: str | None) -> Served: """The bundle a call names, or the fixed one. Never a guess. A one-to-many call that names no bundle is a usage error and not a default: picking one would make the answer's provenance depend on directory order. """ found = self.discovery() served = {entry.bundle_id: entry for entry in found.bundles} if not self.one_to_many: assert self.fixed is not None if self.fixed not in served: raise ToolError( f"the bundle this server was started on is no longer readable: {self.fixed}", code="bundle_unreadable", ) return served[self.fixed] if not bundle_id: raise ToolError( "this server serves several bundles; name one with `bundle_id` " f"({', '.join(sorted(served)) or 'none served'})", code="bundle_id_required", ) if bundle_id in served: return served[bundle_id] for entry in found.unreadable: if entry.path == bundle_id: raise ToolError( f"`{bundle_id}` looks like a bundle and cannot be read as one: {entry.reason}", code="bundle_unreadable", ) raise ToolError( f"no bundle named `{bundle_id}` is served " f"({', '.join(sorted(served)) or 'none served'})", code="bundle_unknown", ) # -------------------------------------------------------------------------- # The card: everything about ONE bundle that a generic consumer needs # -------------------------------------------------------------------------- def card(bundle_root: Path, *, profile: BundleProfile, concept_sample: int = 50) -> dict[str, Any]: """The per-bundle numbers a generic reader needs, DERIVED on demand. This is the half of a generated consumption skill that differs between bundles -- identity, concept count, which conditional fields are written on how many concepts, what the whole bundle costs. Today `okf skill` bakes those numbers into a document, which is what makes the document go stale when the bundle is rebuilt. Derived rather than written into the bundle. Writing a card file into every bundle would move the bytes of all six `examples/*/expected-bundle` trees (23 files compared byte-for-byte) and of the pinned reference bundle, to store something recomputable from the bundle in under a second. A stored card would also be one more artefact that can be stale, which is the defect it was meant to remove. """ from . import skill as okf_skill bundle_id = okf_consume.root_bundle_id_of(bundle_root, profile=profile) concepts = okf_consume.link_parents( [ okf_consume.read_concept( okf_consume.read_path_in_bundle( bundle_root, f"{concept_id}{profile.paths.concept_suffix}" ), bundle_root=bundle_root, root_bundle_id=bundle_id, ) for concept_id in okf_consume.enumerate_concepts(bundle_root, profile=profile) ] ) counts = okf_skill.field_counts(concepts) return { "bundle_id": bundle_id, "ref": okf_consume.bundle_ref(bundle_root, profile=profile), "ref_algorithm": okf_consume.REF_ALGORITHM, "profile": okf_skill.PROFILE_NAME, "concept_count": len(concepts), "concepts": [concept.concept_id for concept in concepts[:concept_sample]], "concepts_truncated": len(concepts) > concept_sample, "source_files": sorted( {concept.source_file for concept in concepts if concept.source_file} ), "conditional_fields": { field: counts.get(field, 0) for field in okf_skill.CONDITIONAL_FIELDS }, "whole_bundle_bytes": okf_skill.whole_bundle_cost(concepts), "budget_unit": okf_consume.BUDGET_UNIT, "default_limit": okf_consume.DEFAULT_LIMIT, } # -------------------------------------------------------------------------- # Tools. Each one has a reason, and the reason is the description a client reads. # -------------------------------------------------------------------------- @dataclass(frozen=True) class Tool: name: str description: str schema: dict[str, Any] _BUNDLE_ARGUMENT = { "bundle_id": { "type": "string", "description": "the bundle to act on; omit on a server started with --bundle", } } def tools(surface: Surface) -> tuple[Tool, ...]: """The minimum set that answers the questions a bundle exists to answer. `okf_list` only on a server that serves more than one: on a one-to-one server there is nothing to list, and a tool that always returns the same single row invites a client to treat discovery as available when the deployment does not have it. """ bundle = _BUNDLE_ARGUMENT if surface.one_to_many else {} listing = ( Tool( "okf_list", "Every OKF bundle this server can currently reach, with its content " "identity and concept count. Re-read from disk on every call, so a " "bundle added, removed or rebuilt since the last call is reflected " "without restarting anything. Exists because a client that cannot " "discover bundles must be told their names out of band, which is the " "configuration this shape is meant to remove.", {"type": "object", "properties": {}, "additionalProperties": False}, ), ) common = ( Tool( "okf_describe", "What one bundle is: its id, its content identity, how many concepts " "it holds, which source documents it was built from, and which " "conditionally-written fields are present on how many concepts. " "Read it BEFORE asking, so the question can be put into the " "bundle's own words. On a multi-bundle server, omitting `bundle_id` " "describes every served bundle, as `okf_ask` does. " "Exists because an answer must be attributable -- a claim from a " "bundle whose identity the caller cannot state is a claim with no " "provenance -- and because a reader needs the denominators before it " "can read an absence.", { "type": "object", "properties": dict(bundle), "additionalProperties": False, }, ), Tool( "okf_ask", "One question, one bounded payload of excerpts, each carrying its " "bundle id, concept id, title and provenance locators, plus what was " "withheld and why. This is the library's only reading direction and " "it calls no model. On a multi-bundle server, omitting `bundle_id` " "asks every served bundle and splits the budget between them. " "ASK IT MORE THAN ONCE: one call answers one wording of one " "sub-question, and `withheld.nearest` names the best-ranked " "concepts that just missed, with their titles -- if one of those is " "what you wanted, ask again in that concept's own words, or fetch " "it by name. Exists " "because handing a client the whole bundle is not an answer, and " "letting it choose files by name is the enumeration the consumption " "contract forbids.", { "type": "object", "properties": { "question": {"type": "string", "description": "the question, in prose"}, **bundle, "k": { "type": "integer", "description": f"how many concepts to consider (default {DEFAULT_K})", }, "limit": {"type": "integer", "description": "payload budget in utf-8 bytes"}, }, "required": ["question"], "additionalProperties": False, }, ), Tool( "okf_fetch", "One named concept, verbatim, with its frontmatter and its source " "locators. Exists because a ranked payload is a SELECTION: an arm " "that has been told a concept id -- by `okf_ask`, by a parent " "pointer, or by a citation it is checking -- needs the bytes " "themselves, and must not have to guess them from an excerpt.", { "type": "object", "properties": { "concept_id": { "type": "string", "description": "a bundle-relative concept id, as `okf_ask` reports it", }, **bundle, }, "required": ["concept_id"], "additionalProperties": False, }, ), ) return (listing + common) if surface.one_to_many else common def call_list(surface: Surface, _arguments: Mapping[str, Any]) -> dict[str, Any]: found = surface.discovery() entries: list[dict[str, Any]] = [] for served in found.bundles: entries.append( { "bundle_id": served.bundle_id, "ref": okf_consume.bundle_ref(served.root, profile=surface.profile), "concept_count": len( okf_consume.enumerate_concepts(served.root, profile=surface.profile) ), "directory": served.root.name, } ) return { "bundles": entries, "unreadable": [ {"directory": entry.path, "reason": entry.reason} for entry in found.unreadable ], "shape": "one-to-many" if surface.one_to_many else "one-to-one", } def call_describe(surface: Surface, arguments: Mapping[str, Any]) -> dict[str, Any]: """One bundle's card, or every served bundle's when none is named. It REFUSED in the second position until 2026-09-20, where `okf_ask` in the same position fans out -- so the one tool a caller is told to read FIRST was the one that required a name it did not have yet. A tool that refuses the call its sibling accepts is a shape a client must be told out of band, which is the configuration this server exists to remove. The named call's shape is UNCHANGED: a caller that passes `bundle_id`, and every one-to-one server, gets exactly the card they always got. The fan-out shape is new where the old behaviour was an error, so there is no caller whose bytes move. """ named = _string(arguments, "bundle_id") if named or not surface.one_to_many: served = surface.resolve(named) return card(served.root, profile=surface.profile) found = surface.discovery() if not found.bundles: raise ToolError("no bundle is served under the given roots", code="bundle_none_served") return { "asked": [served.bundle_id for served in found.bundles], "cards": [card(served.root, profile=surface.profile) for served in found.bundles], } def call_ask(surface: Surface, arguments: Mapping[str, Any]) -> dict[str, Any]: question = _string(arguments, "question") if not question: raise ToolError("`question` is required and may not be empty", code="question_missing") k = int(arguments.get("k") or DEFAULT_K) limit = int(arguments.get("limit") or okf_consume.DEFAULT_LIMIT) named = _string(arguments, "bundle_id") if named or not surface.one_to_many: targets = [surface.resolve(named)] else: targets = list(surface.discovery().bundles) if not targets: raise ToolError("no bundle is served under the given roots", code="bundle_none_served") share = max(1, limit // len(targets)) if share < okf_consume.DEFAULT_LIMIT // 100: raise ToolError( f"the budget splits to {share} bytes across {len(targets)} bundles, which " "cannot carry an excerpt; name one bundle or raise `limit`", code="budget_too_thin", ) answers = [] for served in targets: try: payload = okf_consume.build_payload( served.root, question=question, k=k, limit=share, profile=surface.profile ) except okf_consume.ConsumeError as error: raise ToolError( f"{served.bundle_id}: {error}", code=getattr(error, "code", "consume_refused") ) from error answers.append({"bundle_id": served.bundle_id, "payload": payload}) return { "question": question, "asked": [served.bundle_id for served in targets], "budget_per_bundle": share, "answers": answers, } def call_fetch(surface: Surface, arguments: Mapping[str, Any]) -> dict[str, Any]: concept_id = _string(arguments, "concept_id") if not concept_id: raise ToolError("`concept_id` is required", code="concept_id_missing") served = surface.resolve(_string(arguments, "bundle_id")) known = okf_consume.enumerate_concepts(served.root, profile=surface.profile) if concept_id not in known: raise ToolError( f"`{concept_id}` is not a concept the bundle's index names", code="concept_unknown", ) suffix = surface.profile.paths.concept_suffix # The SECOND of the two independent checks, and since the read paths were # unified it is the same one `okf_ask` and `okf_describe` make. Left as its # own call rather than folded into the index check above: a defect in one # of the two is survivable only while the other is still asked. path = okf_consume.read_path_in_bundle(served.root, f"{concept_id}{suffix}") size = path.stat().st_size if size > MAX_CONCEPT_BYTES: raise ToolError( f"`{concept_id}` is {size} bytes, above this server's {MAX_CONCEPT_BYTES}-byte " "ceiling for one concept; it is refused whole rather than truncated", code="concept_too_large", ) concept = okf_consume.read_concept( path, bundle_root=served.root, root_bundle_id=okf_consume.root_bundle_id_of(served.root, profile=surface.profile), ) return { "bundle_id": concept.bundle_id, "ref": okf_consume.bundle_ref(served.root, profile=surface.profile), "concept": { "concept_id": concept.concept_id, "title": concept.title, "sha256": concept.sha256, "adjudication": concept.adjudication, "req_number": concept.req_number, "source_file": concept.source_file, "sources": [dict(entry) for entry in concept.sources], "locators": dict(concept.locators), "text": concept.body, }, } def _string(arguments: Mapping[str, Any], key: str) -> str: value = arguments.get(key) if value is None: return "" if not isinstance(value, str): raise ToolError( f"`{key}` must be a string, not {type(value).__name__}", code="argument_type" ) return value HANDLERS = { "okf_list": call_list, "okf_describe": call_describe, "okf_ask": call_ask, "okf_fetch": call_fetch, } # -------------------------------------------------------------------------- # The protocol: four methods, newline-delimited JSON-RPC 2.0 over stdio # -------------------------------------------------------------------------- METHOD_NOT_FOUND = -32601 INVALID_PARAMS = -32602 INTERNAL_ERROR = -32603 def _tool_result(payload: Mapping[str, Any]) -> dict[str, Any]: """Both forms, on purpose. `structuredContent` is what a client with a schema reads; the text block is what one without a schema reads, and a client that got only the first would see an empty message. The text is the SAME object, serialised -- two renderings of one answer, never two answers. """ text = json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=False) return { "content": [{"type": "text", "text": text}], "structuredContent": dict(payload), "isError": False, } def _tool_refusal(message: str, code: str) -> dict[str, Any]: return { "content": [{"type": "text", "text": f"refused ({code}): {message}"}], "isError": True, } def handle(surface: Surface, method: str, params: Mapping[str, Any]) -> dict[str, Any]: """One request to one result. Raises `ToolError` only through the envelope.""" if method == "initialize": return { "protocolVersion": PROTOCOL_VERSION, "capabilities": {"tools": {"listChanged": False}}, "serverInfo": {"name": SERVER_NAME, "version": _version()}, "instructions": SERVER_INSTRUCTIONS, } if method == "ping": return {} if method == "tools/list": return { "tools": [ {"name": tool.name, "description": tool.description, "inputSchema": tool.schema} for tool in tools(surface) ] } if method == "tools/call": name = params.get("name") arguments = params.get("arguments") or {} if not isinstance(arguments, Mapping): return _tool_refusal("`arguments` must be an object", "argument_type") available = {tool.name for tool in tools(surface)} if not isinstance(name, str) or name not in available: return _tool_refusal( f"no tool named {name!r} on this server ({', '.join(sorted(available))})", "tool_unknown", ) try: return _tool_result(HANDLERS[name](surface, arguments)) except ToolError as error: return _tool_refusal(str(error), error.code) except okf_consume.ConsumeError as error: return _tool_refusal(str(error), getattr(error, "code", "consume_refused")) except SourceError as error: return _tool_refusal(str(error), getattr(error, "code", "path_escape")) # Broad on purpose: a traceback on stdout would break the framing, and # a server that dies on one bad argument takes every other bundle with # it. The refusal is still loud, and it still carries a code. except Exception as error: return _tool_refusal(f"{type(error).__name__}: {error}", "tool_failed") raise LookupError(method) def _version() -> str: from . import __version__ return __version__ def serve(surface: Surface, *, stdin: TextIO, stdout: TextIO) -> int: """Read requests until stdin closes. One JSON object per line, both ways.""" for line in stdin: line = line.strip() if not line: continue try: message = json.loads(line) except json.JSONDecodeError: continue # unframeable input: there is no id to answer it under if not isinstance(message, dict): continue method = str(message.get("method", "")) identifier = message.get("id") params = message.get("params") or {} if not isinstance(params, Mapping): params = {} if identifier is None: continue # a notification: acknowledged by doing nothing try: result: dict[str, Any] = { "jsonrpc": "2.0", "id": identifier, "result": handle(surface, method, params), } except LookupError: result = { "jsonrpc": "2.0", "id": identifier, "error": {"code": METHOD_NOT_FOUND, "message": f"no method {method!r}"}, } except Exception as error: result = { "jsonrpc": "2.0", "id": identifier, "error": { "code": INTERNAL_ERROR, "message": f"{type(error).__name__}: {error}", }, } stdout.write(json.dumps(result, ensure_ascii=False) + "\n") stdout.flush() return 0 def build_surface( *, bundle: Path | None, roots: Sequence[Path], profile: BundleProfile = okf_consume.DEFAULT_PROFILE, ) -> Surface: if bundle is not None: index = bundle / "index.md" if not index.is_file(): raise ToolError( f"{bundle} carries no index.md, so it is not a bundle", code="not_a_bundle" ) bundle_id = _declared_bundle_id(index) if not bundle_id: raise ToolError(f"{index} declares no bundle_id", code="not_a_bundle") return Surface((bundle.resolve(),), bundle_id, profile) if not roots: raise ToolError("give either --bundle or at least one --root", code="no_target") return Surface(tuple(root.resolve() for root in roots), None, profile) def parse_args(argv: Sequence[str] | None) -> argparse.Namespace: parser = argparse.ArgumentParser( prog="okf mcp", description=( "Serve OKF bundles over the Model Context Protocol on stdio. " "`--bundle` serves one bundle and takes no bundle argument on its " "tools; `--root` serves every bundle found under the given " "directories and knows none of them by name." ), ) parser.add_argument("--bundle", type=Path, help="serve exactly this bundle") parser.add_argument( "--root", type=Path, action="append", default=[], help="serve every bundle under this directory (repeatable)", ) return parser.parse_args(list(argv) if argv is not None else None) def main(argv: Sequence[str] | None = None) -> int: args = parse_args(argv) if args.bundle is not None and args.root: print("okf mcp: --bundle and --root are two shapes; give one", file=sys.stderr) return 2 try: surface = build_surface(bundle=args.bundle, roots=args.root) except ToolError as error: print(f"okf mcp: refused ({error.code}): {error}", file=sys.stderr) return 2 # Line-buffered both ways: a client blocks on our answer, and a block # buffer would hold it until the buffer filled or the process exited. if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(line_buffering=True) return serve(surface, stdin=sys.stdin, stdout=sys.stdout) if __name__ == "__main__": raise SystemExit(main())