char-fork/llama.cpp

git clone https://git.t4t.associates/char-fork/llama.cpp

Xuan-Son Nguyenserver: support MCP stdio (#26062)20455a4ad

char/aliceai-t5-moe
brancheschar/aliceai-t5-moeall refs →
4.2 KiB164 linesraw
1#!/usr/bin/env python3
2"""
3Minimal MCP server for testing.
4Implements JSON-RPC 2.0 over stdio (line-delimited JSON).
5"""
6import json
7import sys
8import os
9
10# Ensure we use python3 from the current environment
11if sys.platform == "win32":
12    # On Windows, we need to use the same python interpreter
13    pass
14
15TOOLS = [
16    {
17        "name": "echo",
18        "description": "Echo back the input message",
19        "inputSchema": {
20            "type": "object",
21            "properties": {
22                "message": {"type": "string", "description": "Message to echo"}
23            },
24            "required": ["message"]
25        }
26    },
27    {
28        "name": "add",
29        "description": "Add two numbers",
30        "inputSchema": {
31            "type": "object",
32            "properties": {
33                "a": {"type": "number"},
34                "b": {"type": "number"}
35            },
36            "required": ["a", "b"]
37        }
38    },
39    {
40        "name": "fail_once",
41        "description": "Fails on first call, succeeds on subsequent calls",
42        "inputSchema": {
43            "type": "object",
44            "properties": {}
45        }
46    }
47]
48
49_state = {"fail_once_called": False}
50
51def handle_initialize(params, req_id):
52    return {
53        "jsonrpc": "2.0",
54        "id": req_id,
55        "result": {
56            "protocolVersion": "2024-11-05",
57            "capabilities": {"tools": {}},
58            "serverInfo": {"name": "echo-test", "version": "1.0"}
59        }
60    }
61
62def handle_tools_list(params, req_id):
63    return {
64        "jsonrpc": "2.0",
65        "id": req_id,
66        "result": {"tools": TOOLS}
67    }
68
69def handle_tools_call(params, req_id):
70    tool_name = params.get("name")
71    arguments = params.get("arguments", {})
72
73    if tool_name == "echo":
74        message = arguments.get("message", "")
75        return {
76            "jsonrpc": "2.0",
77            "id": req_id,
78            "result": {
79                "content": [{"type": "text", "text": f"echo: {message}"}]
80            }
81        }
82    elif tool_name == "add":
83        a = arguments.get("a", 0)
84        b = arguments.get("b", 0)
85        return {
86            "jsonrpc": "2.0",
87            "id": req_id,
88            "result": {
89                "content": [{"type": "text", "text": str(a + b)}]
90            }
91        }
92    elif tool_name == "fail_once":
93        if not _state["fail_once_called"]:
94            _state["fail_once_called"] = True
95            return {
96                "jsonrpc": "2.0",
97                "id": req_id,
98                "error": {"code": -32000, "message": "transient error"}
99            }
100        return {
101            "jsonrpc": "2.0",
102            "id": req_id,
103            "result": {
104                "content": [{"type": "text", "text": "ok"}]
105            }
106        }
107    else:
108        return {
109            "jsonrpc": "2.0",
110            "id": req_id,
111            "error": {"code": -32602, "message": f"Unknown tool: {tool_name}"}
112        }
113
114def handle_ping(params, req_id):
115    return {
116        "jsonrpc": "2.0",
117        "id": req_id,
118        "result": {}
119    }
120
121HANDLERS = {
122    "initialize": handle_initialize,
123    "tools/list": handle_tools_list,
124    "tools/call": handle_tools_call,
125    "ping": handle_ping,
126}
127
128def main():
129    # Use unbuffered output
130    sys.stdout = os.fdopen(sys.stdout.fileno(), "w", buffering=1)
131    sys.stderr = os.fdopen(sys.stderr.fileno(), "w", buffering=1)
132
133    for line in sys.stdin:
134        line = line.strip()
135        if not line:
136            continue
137        try:
138            request = json.loads(line)
139        except json.JSONDecodeError:
140            continue
141
142        method = request.get("method")
143        req_id = request.get("id")
144        params = request.get("params", {})
145
146        # JSON-RPC 2.0: a message without an id is a notification and must not receive a response
147        if req_id is None:
148            continue
149
150        handler = HANDLERS.get(method)
151        if handler:
152            response = handler(params, req_id)
153        else:
154            response = {
155                "jsonrpc": "2.0",
156                "id": req_id,
157                "error": {"code": -32601, "message": f"Method not found: {method}"}
158            }
159
160        sys.stdout.write(json.dumps(response) + "\n")
161        sys.stdout.flush()
162
163if __name__ == "__main__":
164    main()