okami.mom/albedo

git clone https://git.nkp.pet/okami.mom/albedo

dawndaemon,mcp: cache command catalog of plugins such that after a first load so that they dont block /commands afterd3e3d39

main
branchesmainall refs →
2.5 KiB84 linesraw
1"""Minimal line-delimited JSON-RPC MCP server used by mcp_integration.py."""
2
3import json
4import os
5import sys
6import time
7from pathlib import Path
8
9TOOLS = [
10    {
11        "name": "echo",
12        "description": "Echo one message back",
13        "inputSchema": {
14            "type": "object",
15            "required": ["message"],
16            "properties": {"message": {"type": "string"}},
17        },
18    }
19]
20
21
22# Optional test controls: `delay` holds the handshake for that many seconds
23# and `tools` replaces the catalogue, without changing the server's config.
24CONTROL = Path(os.environ.get("FAKE_MCP_CONTROL") or "/nonexistent")
25
26
27def control(name):
28    path = CONTROL / name
29    return path.read_text() if path.exists() else None
30
31
32def reply(id, result=None, error=None):
33    body = {"jsonrpc": "2.0", "id": id}
34    body["error" if error else "result"] = error or result
35    sys.stdout.write(json.dumps(body) + chr(10))
36    sys.stdout.flush()
37
38
39for line in sys.stdin:
40    message = json.loads(line) if line.strip() else {}
41    method, id = message.get("method"), message.get("id")
42    if id is None:
43        continue
44    if method == "initialize":
45        time.sleep(float(control("delay") or 0))
46        reply(
47            id,
48            {
49                "protocolVersion": "2025-06-18",
50                "capabilities": {"tools": {}},
51                "serverInfo": {"name": "fake", "version": "1"},
52            },
53        )
54    elif method == "tools/list":
55        tools = control("tools")
56        reply(id, {"tools": json.loads(tools) if tools else TOOLS})
57    elif method == "tools/call":
58        arguments = message["params"].get("arguments", {})
59        if arguments.get("message") == "fail":
60            reply(id, error={"code": -32000, "message": "the tool exploded"})
61            continue
62        reply(
63            id,
64            {
65                "content": [
66                    {
67                        "type": "text",
68                        "text": json.dumps(
69                            {
70                                "echoed": arguments.get("message"),
71                                "secret": os.environ.get("FAKE_SECRET"),
72                                "ambient": os.environ.get("ALBEDO_MCP_AMBIENT"),
73                            }
74                        ),
75                    }
76                ]
77            },
78        )
79    elif method == "ping":
80        reply(id, {})
81    else:
82        reply(id, error={"code": -32601, "message": "method not found"})
83
84Path(os.environ["FAKE_MCP_CLOSED"]).write_text("closed")