"""MCP server for HDC-resident tools. Speaks the Model Context Protocol over stdio as JSON-RPC 2.0. The tool list is generated from :class:`hdc.agent.ToolRegistry` — that is, from the tool records stored as hypervectors — so what an outside agent sees is exactly what a program can call with the ``TOOL`` instruction. Transports accepted on stdin, both common in the wild: * newline-delimited JSON objects (one request per line); * ``Content-Length: N`` framing followed by N bytes (LSP style). Hypervectors are 10 000-bit integers, which do not survive a JSON number in most clients. They are therefore returned as ``0x``-prefixed hex strings and accepted in either form on input. Run ``python -m hdc.mcp_server --selftest`` to exercise the protocol without a client attached. """ from __future__ import annotations import json import sys from typing import BinaryIO, TextIO from .agent import ToolError, ToolRegistry from .hypervec import DEFAULT_DIM, hvc, popcount from .isa import opcode_table_text from .memory import HDCStorage PROTOCOL_VERSION = "2024-11-05" SERVER_INFO = {"name": "hdcstorage", "version": "1.0.0"} #: JSON-RPC error codes. PARSE_ERROR = -32700 INVALID_REQUEST = -32600 METHOD_NOT_FOUND = -32601 INVALID_PARAMS = -32602 INTERNAL_ERROR = -32603 class MCPServer: """Protocol dispatcher, independent of the transport loop.""" def __init__( self, registry: ToolRegistry | None = None, d: int = DEFAULT_DIM, machine=None, storage: HDCStorage | None = None, ): self.d = d self.storage = storage or HDCStorage(d=d, label="mcp") self.registry = registry or ToolRegistry(storage=self.storage, d=d) self.machine = machine self.initialised = False # ------------------------------------------------------------ resources def resources(self) -> list[dict]: items = [ { "uri": "hdc://isa", "name": "Instruction set", "description": "Opcode table of the hypervector instruction set.", "mimeType": "text/plain", }, { "uri": "hdc://memory", "name": "HDC storage", "description": "Cells currently resident in hyperdimensional storage.", "mimeType": "text/plain", }, { "uri": "hdc://tools", "name": "Tool records", "description": "Agentic tools stored as hypervectors, with capabilities.", "mimeType": "text/plain", }, ] if self.machine is not None and self.machine.prog_len: items.append( { "uri": "hdc://program", "name": "Loaded program", "description": "Disassembly of the program resident in HDC storage.", "mimeType": "text/plain", } ) return items def read_resource(self, uri: str) -> str: if uri == "hdc://isa": return opcode_table_text() if uri == "hdc://memory": return self.storage.dump(limit=64) if uri == "hdc://tools": return self.registry.specs_text() + "\n\n" + self.registry.memory_report() if uri == "hdc://program": if self.machine is None or not self.machine.prog_len: raise KeyError("no program loaded") return self.machine.disasm(0, self.machine.prog_len) raise KeyError("unknown resource: %s" % uri) # --------------------------------------------------------------- tools def _format_result(self, spec, value) -> dict: if spec.returns == "hv": text = "0x%x (%d bits set of %d)" % (int(value), popcount(int(value)), self.d) structured = {"hv": "0x%x" % int(value), "popcount": popcount(int(value))} else: text = str(value) structured = {"value": value} return { "content": [{"type": "text", "text": text}], "structuredContent": structured, "isError": False, } def call_tool(self, name: str, arguments: dict) -> dict: try: spec, confidence = self.registry.resolve(name) if spec is None: return { "content": [ { "type": "text", "text": "unknown tool %r (best similarity %.3f)" % (name, confidence), } ], "isError": True, } args = [] for pname, kind in spec.params: if pname not in arguments: raise ToolError("missing argument %r" % pname) value = arguments[pname] if isinstance(value, str): value = int(value, 0) args.append(int(value)) args = [int(a) for a in args] # Coerce hypervector arguments that arrived as JSON numbers, which # many clients mangle above 2**53 bits of magnitude. value = spec.fn(*args) return self._format_result(spec, value) except Exception as exc: return { "content": [{"type": "text", "text": "%s: %s" % (type(exc).__name__, exc)}], "isError": True, } # ------------------------------------------------------------ dispatch def handle(self, request: dict) -> dict | None: """Handle one JSON-RPC message; returns a response or ``None``.""" if not isinstance(request, dict) or request.get("jsonrpc") != "2.0": return self._error(None, INVALID_REQUEST, "expected a JSON-RPC 2.0 object") method = request.get("method") req_id = request.get("id") params = request.get("params") or {} is_notification = "id" not in request try: if method == "initialize": self.initialised = True requested = params.get("protocolVersion", PROTOCOL_VERSION) result = { "protocolVersion": requested, "capabilities": { "tools": {"listChanged": False}, "resources": {"subscribe": False, "listChanged": False}, }, "serverInfo": SERVER_INFO, "instructions": ( "HDCSTORAGE exposes hyperdimensional storage as tools: " "vectors are hex strings, tools are dispatched by " "similarity, and every call is answered from HDC memory." ), } elif method in ("notifications/initialized", "initialized"): return None elif method == "ping": result = {} elif method == "tools/list": result = {"tools": self.registry.list_tools()} elif method == "tools/call": name = params.get("name") if not isinstance(name, str): return self._error(req_id, INVALID_PARAMS, "tools/call needs a tool name") result = self.call_tool(name, params.get("arguments") or {}) elif method == "resources/list": result = {"resources": self.resources()} elif method == "resources/read": uri = params.get("uri") try: text = self.read_resource(str(uri)) except KeyError as exc: return self._error(req_id, INVALID_PARAMS, str(exc)) result = { "contents": [ {"uri": uri, "mimeType": "text/plain", "text": text} ] } elif method == "shutdown": result = {} else: return self._error(req_id, METHOD_NOT_FOUND, "unknown method %r" % (method,)) except Exception as exc: # pragma: no cover - defensive return self._error(req_id, INTERNAL_ERROR, "%s: %s" % (type(exc).__name__, exc)) if is_notification: return None return {"jsonrpc": "2.0", "id": req_id, "result": result} def _error(self, req_id, code: int, message: str) -> dict: return {"jsonrpc": "2.0", "id": req_id, "error": {"code": code, "message": message}} # ------------------------------------------------------------- framing def serve(self, stdin: TextIO | BinaryIO, stdout: TextIO | BinaryIO) -> None: """Read requests until EOF, writing one response per request.""" binary = False while True: line = stdin.readline() if not line: break if isinstance(line, bytes): binary = True line = line.decode("utf-8", "replace") line = line.strip() if not line: continue payload = line if line.lower().startswith("content-length:"): length = int(line.split(":", 1)[1].strip()) while True: header = stdin.readline() if isinstance(header, bytes): header = header.decode("utf-8", "replace") if header.strip() == "": break raw = stdin.read(length) if isinstance(raw, bytes): raw = raw.decode("utf-8", "replace") payload = raw try: request = json.loads(payload) except json.JSONDecodeError as exc: self._write(stdout, self._error(None, PARSE_ERROR, "invalid JSON: %s" % exc), binary) continue if isinstance(request, list): responses = [r for r in (self.handle(item) for item in request) if r] if responses: self._write(stdout, responses, binary) continue response = self.handle(request) if response is not None: self._write(stdout, response, binary) @staticmethod def _write(stdout, payload, binary: bool) -> None: text = json.dumps(payload) if binary: stdout.write((text + "\n").encode("utf-8")) else: stdout.write(text + "\n") stdout.flush() def selftest(server: MCPServer | None = None) -> str: """Exercise the protocol end to end and return a transcript.""" server = server or MCPServer(d=2000) script = [ {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": PROTOCOL_VERSION, "clientInfo": {"name": "selftest"}}}, {"jsonrpc": "2.0", "method": "notifications/initialized"}, {"jsonrpc": "2.0", "id": 2, "method": "tools/list"}, {"jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": {"name": "hdc.encode_int", "arguments": {"n": 42}}}, ] lines = ["MCP selftest -> %s (d=%d)" % (SERVER_INFO["name"], server.d)] encoded = None for message in script: response = server.handle(message) lines.append("--> " + json.dumps(message)) if response is None: lines.append("<-- (notification, no response)") continue lines.append("<-- " + json.dumps(response)) if message.get("id") == 3 and "result" in response: encoded = response["result"]["structuredContent"].get("hv") if encoded: for name, args in ( ("hdc.decode_int", {"v": encoded}), ("hdc.similarity", {"a": encoded, "b": encoded}), ("hdc.bind", {"a": encoded, "b": encoded}), ("hdc.bundl", {"a": encoded, "b": encoded}), # deliberate typo ("hdc.memory.search", {"v": encoded, "k": 3}), # deliberate paraphrase ("bit.popcount", {"v": encoded}), ): response = server.handle( {"jsonrpc": "2.0", "id": 99, "method": "tools/call", "params": {"name": name, "arguments": args}} ) spec, conf = server.registry.resolve(name) lines.append( "%-22s -> %-16s conf=%.3f %s" % ( name, spec.name if spec else "?", conf, response["result"]["content"][0]["text"][:60], ) ) lines.append(server.read_resource("hdc://tools").splitlines()[0]) return "\n".join(lines) def main(argv: list[str] | None = None) -> int: argv = list(sys.argv[1:] if argv is None else argv) d = DEFAULT_DIM if "--d" in argv: i = argv.index("--d") d = int(argv[i + 1]) server = MCPServer(d=d) if "--selftest" in argv: print(selftest(server)) return 0 print( "hdcstorage MCP server on stdio (d=%d, %d tools). Ctrl-C to stop." % (d, len(server.registry.specs)), file=sys.stderr, ) try: server.serve(sys.stdin, sys.stdout) except KeyboardInterrupt: return 0 return 0 if __name__ == "__main__": raise SystemExit(main())