neuron/cli/neuron_mcp.py

#!/usr/bin/env python3
"""
Neuron MCP server — talk to the local Neuron soul (:7770) from Claude Code.

Stdlib only (no pip deps). stdio transport, newline-delimited JSON-RPC 2.0.
Exposes:
  - neuron_chat(message, agentic?)       -> the soul's reply
  - neuron_search_knowledge(query, limit?) -> lexical knowledge search
  - neuron_search_memory(query, limit?)    -> memory/recall search
"""
import sys, json, urllib.request, urllib.parse

SOUL = "http://127.0.0.1:7770"


def _post(path, payload, timeout=180):
    data = json.dumps(payload).encode()
    req = urllib.request.Request(SOUL + path, data=data,
                                 headers={"Content-Type": "application/json"}, method="POST")
    with urllib.request.urlopen(req, timeout=timeout) as r:
        return json.loads(r.read().decode("utf-8", "replace"))


def _get(path, timeout=30):
    req = urllib.request.Request(SOUL + path, method="GET")
    with urllib.request.urlopen(req, timeout=timeout) as r:
        return r.read().decode("utf-8", "replace")


def neuron_chat(args):
    msg = (args.get("message") or "").strip()
    if not msg:
        return "error: message is required"
    agentic = bool(args.get("agentic", False))
    try:
        resp = _post("/api/chat", {"session_id": "", "message": msg, "agentic": agentic})
    except Exception as e:
        return f"error talking to Neuron (:7770): {e}"
    return resp.get("response") or resp.get("reply") or json.dumps(resp)[:2000]


def _search(path_tmpl, args):
    q = (args.get("query") or "").strip()
    if not q:
        return "error: query is required"
    limit = int(args.get("limit", 5))
    try:
        raw = _get(path_tmpl.format(q=urllib.parse.quote(q), n=limit))
    except Exception as e:
        return f"error searching Neuron: {e}"
    try:
        arr = json.loads(raw)
    except Exception:
        return raw[:2000]
    # The soul returns HTTP 200 with a JSON error object (not a list) when a
    # downstream service is unreachable, e.g. memory recall proxies to :7771.
    if isinstance(arr, dict):
        err = str(arr.get("error", "")).lower()
        if "7771" in err or "connect" in err:
            return ("memory recall is unavailable: the soul's recall backend "
                    "(:7771) isn't running. neuron_chat and "
                    "neuron_search_knowledge still work.")
        return f"error from Neuron: {arr.get('error') or json.dumps(arr)[:500]}"
    if not isinstance(arr, list):
        return str(arr)[:2000]
    if not arr:
        return "no results"
    out = []
    for n in arr[:limit]:
        nid = n.get("id", "")
        content = str(n.get("content", "")).replace("\n", " ")[:300]
        out.append(f"- [{nid}] {content}")
    return "\n".join(out)


def neuron_search_knowledge(args):
    return _search("/api/neuron/knowledge/search?q={q}&limit={n}", args)


def neuron_search_memory(args):
    return _search("/api/memories/recall?query={q}&limit={n}", args)


TOOLS = [
    {"name": "neuron_chat",
     "description": "Send a message to the local Neuron soul and return its reply. Use this to talk to Neuron.",
     "inputSchema": {"type": "object", "properties": {
         "message": {"type": "string", "description": "What to say to Neuron"},
         "agentic": {"type": "boolean", "description": "Use agentic/tool mode (default false)"}},
         "required": ["message"]}},
    {"name": "neuron_search_knowledge",
     "description": "Search Neuron's knowledge base (lexical/keyword match).",
     "inputSchema": {"type": "object", "properties": {
         "query": {"type": "string"}, "limit": {"type": "integer"}}, "required": ["query"]}},
    {"name": "neuron_search_memory",
     "description": "Search what Neuron remembers (memory recall).",
     "inputSchema": {"type": "object", "properties": {
         "query": {"type": "string"}, "limit": {"type": "integer"}}, "required": ["query"]}},
]
HANDLERS = {"neuron_chat": neuron_chat,
            "neuron_search_knowledge": neuron_search_knowledge,
            "neuron_search_memory": neuron_search_memory}


def send(msg):
    sys.stdout.write(json.dumps(msg) + "\n")
    sys.stdout.flush()


def main():
    for line in sys.stdin:
        line = line.strip()
        if not line:
            continue
        try:
            req = json.loads(line)
        except Exception:
            continue
        mid = req.get("id")
        method = req.get("method")
        if method == "initialize":
            pv = (req.get("params") or {}).get("protocolVersion") or "2024-11-05"
            send({"jsonrpc": "2.0", "id": mid, "result": {
                "protocolVersion": pv,
                "capabilities": {"tools": {}},
                "serverInfo": {"name": "neuron", "version": "0.1.0"}}})
        elif method == "notifications/initialized":
            pass
        elif method == "ping":
            send({"jsonrpc": "2.0", "id": mid, "result": {}})
        elif method == "tools/list":
            send({"jsonrpc": "2.0", "id": mid, "result": {"tools": TOOLS}})
        elif method == "tools/call":
            params = req.get("params") or {}
            name = params.get("name")
            args = params.get("arguments") or {}
            fn = HANDLERS.get(name)
            if not fn:
                send({"jsonrpc": "2.0", "id": mid, "result": {
                    "content": [{"type": "text", "text": f"unknown tool: {name}"}], "isError": True}})
            else:
                try:
                    text = fn(args)
                except Exception as e:
                    text = f"error: {e}"
                send({"jsonrpc": "2.0", "id": mid, "result": {
                    "content": [{"type": "text", "text": str(text)}]}})
        elif mid is not None:
            send({"jsonrpc": "2.0", "id": mid,
                  "error": {"code": -32601, "message": f"method not found: {method}"}})


if __name__ == "__main__":
    try:
        main()
    except (BrokenPipeError, KeyboardInterrupt):
        pass