Files
gxde-ai-see/tests/mcp/test_server.py
T
shenmo7192 77a4369cd6 feat(chat): MCP stdio 工具接入
- McpServer:启动进程、initialize、tools/list、tools/call
- McpManager:多服务器、工具名冲突处理、OpenAI function 定义
- ChatClient:请求携带 tools,流式累积 tool_calls,结束后交回上层
- ChatWindow:工具调用循环(最多 5 轮)、scaffold 消息、设置里的 MCP JSON 编辑
- 无头自测 gxde-ai-see-mcptest + tests/mcp/test_server.py 验证通过
2026-10-30 20:22:14 +08:00

46 lines
1.3 KiB
Python

#!/usr/bin/env python3
# SPDX-FileCopyrightText: 2026 GXDE OS Team
# SPDX-License-Identifier: GPL-3.0-or-later
"""Tiny stdio MCP server used by tests/mcptest.cpp."""
import json
import sys
def reply(message, result):
print(json.dumps({"jsonrpc": "2.0", "id": message["id"], "result": result}), flush=True)
for line in sys.stdin:
line = line.strip()
if not line:
continue
message = json.loads(line)
method = message.get("method")
if method == "initialize":
reply(message, {
"protocolVersion": "2024-11-05",
"capabilities": {"tools": {}},
"serverInfo": {"name": "test-server", "version": "0.1"},
})
elif method == "tools/list":
reply(message, {
"tools": [{
"name": "echo",
"description": "Echo the given text back",
"inputSchema": {
"type": "object",
"properties": {"text": {"type": "string"}},
"required": ["text"],
},
}]
})
elif method == "tools/call":
arguments = message.get("params", {}).get("arguments", {})
reply(message, {
"content": [{"type": "text", "text": "echo: " + str(arguments.get("text", ""))}]
})
elif "id" in message:
reply(message, {})