jochen
78817c949e
The node's runtime launches a tools bundle as a process and speaks MCP over stdio to it, so a tool may be written in any language and the transport stays in the runtime (ADR 0039's refusal, kept). Each SDK is the loop and the tool type and nothing else: `@novox/mesh-sdk/stdio` (serveStdio, and serveRegisteredOverStdio for a bundle that already registers the in-process way), python/, go/, rust/ and c/ beside it, each with an example answering `greet` and the seat verb `node-lamp.on`. Skeletons, by the operator's direction: the bare minimum one bundle per language needs to be built and answer, proven here by each example answering tools/list and tools/call on stdio.
78 lines
3.2 KiB
Python
78 lines
3.2 KiB
Python
"""The Novox Mesh SDK for Python — the skeleton (novox/hq ADR 0187).
|
|
|
|
A tools bundle is a process the node's runtime launches and speaks MCP over stdio to: `initialize`,
|
|
`tools/list` once, `tools/call` per call, newline-framed JSON-RPC on stdin and stdout. This module is
|
|
the loop, so a bundle writes its tools and nothing else. A tool named `<seat>.<verb>` is the
|
|
module's implementation of that seat's verb; any other name is the module's own.
|
|
|
|
No transport here: the bus, the subjects and the memberships are the runtime's (ADR 0039, 0175).
|
|
stdout is the protocol; log to stderr.
|
|
"""
|
|
import json
|
|
import sys
|
|
from dataclasses import dataclass, field
|
|
from typing import Any, Callable, Dict, List
|
|
|
|
PROTOCOL = "2025-03-26"
|
|
|
|
|
|
@dataclass
|
|
class Tool:
|
|
name: str
|
|
description: str
|
|
run: Callable[[Dict[str, Any]], Any]
|
|
# The JSON schema of the arguments, or the bare property map the catalogue's modules write.
|
|
input: Dict[str, Any] = field(default_factory=dict)
|
|
|
|
|
|
def as_schema(given: Dict[str, Any]) -> Dict[str, Any]:
|
|
if not given:
|
|
return {"type": "object", "properties": {}}
|
|
if given.get("type") == "object" or "properties" in given:
|
|
return given
|
|
return {"type": "object", "properties": given}
|
|
|
|
|
|
def serve(name: str, tools: List[Tool]) -> None:
|
|
"""Serve these tools to the runtime over stdio until stdin closes."""
|
|
by_name = {t.name: t for t in tools}
|
|
out = sys.stdout
|
|
|
|
def say(message: Dict[str, Any]) -> None:
|
|
out.write(json.dumps(message) + "\n")
|
|
out.flush()
|
|
|
|
for line in sys.stdin:
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
try:
|
|
request = json.loads(line)
|
|
except ValueError:
|
|
continue
|
|
rid = request.get("id")
|
|
method = request.get("method")
|
|
params = request.get("params") or {}
|
|
if method == "initialize":
|
|
say({"jsonrpc": "2.0", "id": rid, "result": {"protocolVersion": PROTOCOL, "capabilities": {"tools": {}}, "serverInfo": {"name": name, "version": "1"}}})
|
|
elif method == "notifications/initialized":
|
|
pass
|
|
elif method == "ping":
|
|
if rid is not None:
|
|
say({"jsonrpc": "2.0", "id": rid, "result": {}})
|
|
elif method == "tools/list":
|
|
say({"jsonrpc": "2.0", "id": rid, "result": {"tools": [
|
|
{"name": t.name, "description": t.description, "inputSchema": as_schema(t.input)} for t in tools]}})
|
|
elif method == "tools/call":
|
|
tool = by_name.get(str(params.get("name", "")))
|
|
if tool is None:
|
|
say({"jsonrpc": "2.0", "id": rid, "error": {"code": -32602, "message": f"{name} has no tool {params.get('name')}"}})
|
|
continue
|
|
try:
|
|
result = tool.run(params.get("arguments") or {})
|
|
say({"jsonrpc": "2.0", "id": rid, "result": {"content": [{"type": "text", "text": json.dumps(result)}]}})
|
|
except Exception as err: # the tool failed: a tool error, not a protocol one
|
|
say({"jsonrpc": "2.0", "id": rid, "result": {"content": [{"type": "text", "text": str(err)}], "isError": True}})
|
|
elif rid is not None:
|
|
say({"jsonrpc": "2.0", "id": rid, "error": {"code": -32601, "message": f"no {method}"}})
|