105 lines
3.7 KiB
Python
105 lines
3.7 KiB
Python
|
"""本地 MCP 测试服务器（Streamable HTTP / JSON-RPC POST）。
|
||
|
|
|
||
|
|
用法：python scripts/test_mcp_server.py [端口] 默认 8765
|
||
|
|
提供两个工具：
|
||
|
|
- get_current_time：返回当前时间（无参数）
|
||
|
|
- echo_text：原样返回 text 参数
|
||
|
|
"""
|
||
|
|
|
||
|
|
import json
|
||
|
|
import sys
|
||
|
|
import time
|
||
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
||
|
|
|
||
|
|
|
||
|
|
class McpHandler(BaseHTTPRequestHandler):
|
||
|
|
def do_GET(self):
|
||
|
|
# 供健康检查/就绪探测使用
|
||
|
|
self._json({"ok": True})
|
||
|
|
|
||
|
|
def do_POST(self):
|
||
|
|
length = int(self.headers.get("Content-Length", 0))
|
||
|
|
body = self.rfile.read(length).decode("utf-8")
|
||
|
|
try:
|
||
|
|
req = json.loads(body)
|
||
|
|
except Exception:
|
||
|
|
self._json(
|
||
|
|
{
|
||
|
|
"jsonrpc": "2.0",
|
||
|
|
"id": None,
|
||
|
|
"error": {"code": -32700, "message": "Parse error"},
|
||
|
|
}
|
||
|
|
)
|
||
|
|
return
|
||
|
|
method = req.get("method")
|
||
|
|
rid = req.get("id")
|
||
|
|
if method == "initialize":
|
||
|
|
result = {
|
||
|
|
"protocolVersion": "2025-03-26",
|
||
|
|
"capabilities": {"tools": {}},
|
||
|
|
"serverInfo": {"name": "test-mcp-server", "version": "1.0.0"},
|
||
|
|
}
|
||
|
|
self._json({"jsonrpc": "2.0", "id": rid, "result": result})
|
||
|
|
elif method == "tools/list":
|
||
|
|
result = {
|
||
|
|
"tools": [
|
||
|
|
{
|
||
|
|
"name": "get_current_time",
|
||
|
|
"description": "获取当前时间",
|
||
|
|
"inputSchema": {"type": "object", "properties": {}},
|
||
|
|
},
|
||
|
|
{
|
||
|
|
"name": "echo_text",
|
||
|
|
"description": "原样返回输入的文本",
|
||
|
|
"inputSchema": {
|
||
|
|
"type": "object",
|
||
|
|
"properties": {"text": {"type": "string"}},
|
||
|
|
"required": ["text"],
|
||
|
|
},
|
||
|
|
},
|
||
|
|
]
|
||
|
|
}
|
||
|
|
self._json({"jsonrpc": "2.0", "id": rid, "result": result})
|
||
|
|
elif method == "tools/call":
|
||
|
|
params = req.get("params", {})
|
||
|
|
name = params.get("name")
|
||
|
|
args = params.get("arguments", {})
|
||
|
|
if name == "get_current_time":
|
||
|
|
content = [{"type": "text", "text": time.strftime("%Y-%m-%d %H:%M:%S")}]
|
||
|
|
elif name == "echo_text":
|
||
|
|
content = [{"type": "text", "text": str(args.get("text", ""))}]
|
||
|
|
else:
|
||
|
|
self._json(
|
||
|
|
{
|
||
|
|
"jsonrpc": "2.0",
|
||
|
|
"id": rid,
|
||
|
|
"error": {"code": -32602, "message": f"unknown tool {name}"},
|
||
|
|
}
|
||
|
|
)
|
||
|
|
return
|
||
|
|
self._json({"jsonrpc": "2.0", "id": rid, "result": {"content": content}})
|
||
|
|
else:
|
||
|
|
self._json(
|
||
|
|
{
|
||
|
|
"jsonrpc": "2.0",
|
||
|
|
"id": rid,
|
||
|
|
"error": {"code": -32601, "message": f"unknown method {method}"},
|
||
|
|
}
|
||
|
|
)
|
||
|
|
|
||
|
|
def _json(self, obj):
|
||
|
|
data = json.dumps(obj).encode("utf-8")
|
||
|
|
self.send_response(200)
|
||
|
|
self.send_header("Content-Type", "application/json")
|
||
|
|
self.send_header("Content-Length", str(len(data)))
|
||
|
|
self.end_headers()
|
||
|
|
self.wfile.write(data)
|
||
|
|
|
||
|
|
def log_message(self, *args):
|
||
|
|
pass
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
port = int(sys.argv[1]) if len(sys.argv) > 1 else 8765
|
||
|
|
print(f"test-mcp-server listening on http://127.0.0.1:{port}/mcp")
|
||
|
|
HTTPServer(("127.0.0.1", port), McpHandler).serve_forever()
|