open/llm-ingestion-okf
1
0
Fork 0
Code Issues Releases 4 Activity

feat(mcp): serve OKF bundles over MCP in two shapes, plus the generic skill

Browse source 
The eval was written RED at `5f1772e` with no server in the tree. This is the
capability it was written against.

`okf mcp --bundle <dir>` serves exactly one bundle, whose tools take no bundle
argument. `okf mcp --root <dir>` (repeatable) serves every bundle under the
roots and knows NONE of them by name. Four tools -- `okf_list`,
`okf_describe`, `okf_ask`, `okf_fetch` -- each carrying its reason in the
description a client actually reads.

Gate today: 1 (7/7) - 2 (83/181) - 3 (4/4) - 4 (9/9) - 5 (3/3) - 6 (6/6),
`GATE RED: rows 2`, exit 1.

THE PROTOCOL IS STDLIB, AND THAT IS THE PACKAGING INVARIANT KEPT RATHER THAN
A TASTE. An MCP SDK would be this package's second runtime dependency on the
DEFAULT install path, for four JSON-RPC methods and a newline framing, and
`test_the_only_runtime_dependency_is_the_security_boundary` pins that list
literally. Chosen hand-written because the surface needed is `initialize`,
`notifications/initialized`, `tools/list` and `tools/call`; `uv.lock` is
untouched.

NOTHING IS CACHED ACROSS CALLS, and row 3 is why. Every call re-walks the
roots and recomputes `bundle_ref`, so a bundle added, removed or rebuilt while
the process runs is seen by the next call with no restart, no configuration
edit and no code change -- 9 of 9 discovery checks over three bundles written
while the server was serving. The cost is paid per call and is published
rather than hidden: 0.75 s for the identity of a 2 756-concept bundle, 5.6 s
for one ask, 4 min 13 s for row 2's full run over four bundles.

CONTAINMENT IS TWO INDEPENDENT CHECKS: the bundle's own index must name the
concept, AND `connectors.safe_resolve` must place it inside the bundle. A
mutant removing either one alone still refuses -- with a DIFFERENT code, which
row 6 asserts by name -- and one removing both is killed. Row 6 declares a
code set per case because its first run had the 10 MB concept refused as
`concept_unknown`: the fixture had not named the file in the index, so the
size ceiling never ran and the row was green for a reason unrelated to the
attack.

`okf card <bundle>` and `okf skill --generic` are the one-to-many skill
candidate. The card is DERIVED on every run and never written into the bundle:
storing it would move the bytes of all six `examples/*/expected-bundle` trees
(23 files compared byte-for-byte) and of the pinned reference bundle, to keep
something recomputable in under a second, and a stored card is one more
artefact that can disagree with what is beside it. Measured here rather than
taken from the order: two per-bundle skills are identical on 280 of 312 and
310 lines; the 62 that differ are identity, concept count, the
conditional-field table, the whole-bundle cost and the breaking point. The
generic skill carries none of them, and `render_generic()` takes no argument,
so there is no bundle it could have read.

Row 2 decomposes into three numbers and the middle one is the finding: 99 of
181 (bundle, anchor) pairs are present in the bundles at all, 83 of those 99
were reached, and 0 of 83 were met by `okf_fetch` on the anchor as a concept
id. The set's anchors and this library's concept ids are different
vocabularies, so every pair met was met through the ranker -- 83 is a FLOOR on
the ceiling, never the ceiling.

13 mutants in a scratch copy, never in the working tree: 12 killed, 1 survived
with its mechanism printed, 0 errors, control green first. Suite 2323 passed,
2 skipped. The architecture choice between the two shapes is the OPERATOR's;
these rows are its input. Report: docs/2026-09-20-mcp-to-varianter.md.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Kjell Tore Guttormsen 2026-09-20 10:25:55 +02:00
commit df5a1183c9
Signed by: ktg
SSH key fingerprint: SHA256:JakMjO6FTBBzN0Bhfj9saOoEjaFxlSdYuZQQpM/lF9Q
10 changed files with 1823 additions and 59 deletions
Download patch file Download diff file Expand all files Collapse all files

737
src/llm_ingestion_okf/mcp_server.py Normal file
View file

@ -0,0 +1,737 @@
"""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.
"""
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 .connectors import safe_resolve
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"
#: 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(
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. "
"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. 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]:
served = surface.resolve(_string(arguments, "bundle_id"))
return card(served.root, profile=surface.profile)
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
try:
path = safe_resolve(served.root, f"{concept_id}{suffix}")
except SourceError as error:
raise ToolError(str(error), code="path_escape") from error
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": (
"Bundles are read-only. Ask `okf_ask` a question in prose rather "
"than fetching concepts by name: every excerpt it returns carries "
"the bundle id and concept id a claim must be attributed to, and "
"the payload states what it withheld and why."
),
}
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())