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