| """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"} |
|
|
| |
| 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 |
|
|
| |
| 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) |
|
|
| |
| 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] |
| |
| |
| 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, |
| } |
|
|
| |
| 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: |
| 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}} |
|
|
| |
| 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}), |
| ("hdc.memory.search", {"v": encoded, "k": 3}), |
| ("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()) |
|
|
