raisfast/scripts/agents/mcp_echo_server.py

#!/usr/bin/env python3
"""Tiny MCP stdio server (JSON-RPC over newline-delimited lines) for smoke tests.

Implements initialize / notifications/initialized / tools/list / tools/call.
Speaks only to the minimal subset raisfast's MCP client uses.
"""
import json
import sys


def reply(msg):
    if msg is None:
        return
    if isinstance(msg, dict):
        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 json.JSONDecodeError:
            continue
        method = req.get("method", "")
        rid = req.get("id")
        if method == "initialize":
            reply({"jsonrpc": "2.0", "id": rid, "result": {
                "protocolVersion": req.get("params", {}).get("protocolVersion", "2024-11-05"),
                "capabilities": {"tools": {}},
                "serverInfo": {"name": "echo", "version": "1.0"},
            }})
        elif method == "notifications/initialized":
            continue
        elif method == "tools/list":
            reply({"jsonrpc": "2.0", "id": rid, "result": {"tools": [
                {"name": "echo",
                 "description": "Echoes back the `msg` argument",
                 "inputSchema": {"type": "object",
                                 "properties": {"msg": {"type": "string"}},
                                 "required": ["msg"]}},
            ]}})
        elif method == "tools/call":
            params = req.get("params", {})
            name = params.get("name")
            args = params.get("arguments", {}) or {}
            if name == "echo":
                text = "echo:" + str(args.get("msg", ""))
            else:
                text = f"unknown tool {name}"
            reply({"jsonrpc": "2.0", "id": rid, "result": {
                "content": [{"type": "text", "text": text}],
                "isError": False,
            }})
        else:
            reply({"jsonrpc": "2.0", "id": rid,
                   "error": {"code": -32601, "message": "not implemented"}})


if __name__ == "__main__":
    main()