ben/mgmt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 重构项目脚本结构和文档

Browse Source 
docs: 添加API和部署文档
refactor: 重新组织脚本目录结构
fix: 修复Nomad配置中的region设置
chore: 移除过期节点相关配置
test: 更新MCP服务器测试脚本
build: 更新Makefile以适配新脚本路径
This commit is contained in:
Houzhong Xu
2025-10-01 02:08:58 +00:00
parent 7ea230b072
commit e5aa00d6f9
119 changed files with 7847 additions and 203 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
tests/mcp_servers/test_direct_search.sh
View File

@@ -1,32 +0,0 @@
#!/bin/bash
echo "直接测试search_documents方法..."
# 创建一个简单的Python脚本来测试search_documents方法
ssh ben@dev1 "cd /home/ben/qdrant && source venv/bin/activate && python3 -c \"
import asyncio
import json
import sys
sys.path.append('/home/ben/qdrant')
from qdrant_ollama_mcp_server import QdrantOllamaMCPServer
async def test_search():
server = QdrantOllamaMCPServer()
# 测试search_documents方法
params = {
'query': '人工智能',
'limit': 3
}
try:
result = await server._search_documents(params)
print('搜索结果:', json.dumps(result, indent=2, ensure_ascii=False))
except Exception as e:
print('搜索错误:', str(e))
import traceback
traceback.print_exc()
asyncio.run(test_search())
\""

61
tests/mcp_servers/test_local_mcp_servers.sh
View File

@@ -1,61 +0,0 @@
#!/bin/bash
# 测试当前环境中的MCP服务器
echo "测试当前环境中的MCP服务器..."
# 检查当前环境中是否有MCP配置
echo "检查MCP配置..."
if [ -f "/root/.mcp/mcp_settings.json" ]; then
echo "找到MCP配置文件: /root/.mcp/mcp_settings.json"
cat /root/.mcp/mcp_settings.json
else
echo "未找到MCP配置文件: /root/.mcp/mcp_settings.json"
fi
echo ""
echo "检查.kilocode/mcp.json..."
if [ -f "/root/mgmt/.kilocode/mcp.json" ]; then
echo "找到MCP配置文件: /root/mgmt/.kilocode/mcp.json"
cat /root/mgmt/.kilocode/mcp.json
else
echo "未找到MCP配置文件: /root/mgmt/.kilocode/mcp.json"
fi
echo ""
echo "检查是否有可用的MCP服务器..."
# 检查context7服务器
echo "测试context7服务器..."
echo '{"jsonrpc":"2.0","id":1,"method":"tools/list"}' | nc localhost 8080 2>/dev/null || echo "context7服务器未在本地运行"
# 检查qdrant服务器
echo "测试qdrant服务器..."
if [ -f "/root/mgmt/qdrant_mcp_server.py" ]; then
echo "找到qdrant服务器脚本: /root/mgmt/qdrant_mcp_server.py"
# 尝试直接运行服务器并测试
echo '{"jsonrpc":"2.0","id":1,"method":"tools/list"}' | python3 /root/mgmt/qdrant_mcp_server.py 2>/dev/null || echo "qdrant服务器无法直接运行"
else
echo "未找到qdrant服务器脚本"
fi
# 检查qdrant-ollama服务器
echo "测试qdrant-ollama服务器..."
if [ -f "/root/mgmt/qdrant_ollama_mcp_server.py" ]; then
echo "找到qdrant-ollama服务器脚本: /root/mgmt/qdrant_ollama_mcp_server.py"
# 尝试直接运行服务器并测试
echo '{"jsonrpc":"2.0","id":1,"method":"tools/list"}' | python3 /root/mgmt/qdrant_ollama_mcp_server.py 2>/dev/null || echo "qdrant-ollama服务器无法直接运行"
else
echo "未找到qdrant-ollama服务器脚本"
fi
echo ""
echo "检查环境变量..."
echo "QDRANT_URL: ${QDRANT_URL:-未设置}"
echo "QDRANT_API_KEY: ${QDRANT_API_KEY:-未设置}"
echo "OLLAMA_URL: ${OLLAMA_URL:-未设置}"
echo "OLLAMA_MODEL: ${OLLAMA_MODEL:-未设置}"
echo "COLLECTION_NAME: ${COLLECTION_NAME:-未设置}"
echo ""
echo "测试完成。"

21
tests/mcp_servers/test_mcp_interface.sh
View File

@@ -1,21 +0,0 @@
#!/bin/bash
# 测试MCP服务器在实际MCP接口中的调用
echo "测试Qdrant MCP服务器..."
echo '{"jsonrpc":"2.0","id":1,"method":"tools/list"}' | ssh ben@dev1 "cd /home/ben/qdrant && source venv/bin/activate && python qdrant_mcp_server.py"
echo ""
echo "测试Qdrant-Ollama MCP服务器..."
echo '{"jsonrpc":"2.0","id":1,"method":"tools/list"}' | ssh ben@dev1 "cd /home/ben/qdrant && source venv/bin/activate && ./start_mcp_server.sh"
echo ""
echo "测试Qdrant MCP服务器的搜索功能..."
echo '{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"qdrant_search","arguments":{"query":"测试查询","limit":3}}}' | ssh ben@dev1 "cd /home/ben/qdrant && source venv/bin/activate && python qdrant_mcp_server.py"
echo ""
echo "测试Qdrant-Ollama MCP服务器的搜索功能..."
echo '{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"qdrant_search","arguments":{"query":"测试查询","limit":3}}}' | ssh ben@dev1 "cd /home/ben/qdrant && source venv/bin/activate && ./start_mcp_server.sh"
echo ""
echo "测试完成。"

15
tests/mcp_servers/test_mcp_search_final.sh
View File

@@ -1,15 +0,0 @@
#!/bin/bash
echo "测试通过MCP接口调用search_documents工具..."
# 先添加一个文档
echo "添加测试文档..."
ssh ben@dev1 "cd /home/ben/qdrant && source venv/bin/activate && echo '{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"tools/call\",\"params\":{\"name\":\"add_document\",\"arguments\":{\"text\":\"机器学习是人工智能的一个子领域，专注于开发能够从数据中学习的算法。\",\"metadata\":{\"source\":\"test\",\"topic\":\"ML\"}}}}' | ./start_mcp_server.sh"
echo ""
echo "通过MCP接口搜索文档..."
# 测试search_documents工具（不带filter参数）
ssh ben@dev1 "cd /home/ben/qdrant && source venv/bin/activate && echo '{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"tools/call\",\"params\":{\"name\":\"search_documents\",\"arguments\":{\"query\":\"机器学习\",\"limit\":3}}}' | ./start_mcp_server.sh"
echo ""
echo "测试完成。"

13
tests/mcp_servers/test_mcp_servers.sh
View File

@@ -1,13 +0,0 @@
#!/bin/bash
# 测试MCP服务器脚本
echo "测试Qdrant MCP服务器..."
echo '{"jsonrpc":"2.0","id":1,"method":"initialize"}' | ssh ben@dev1 "cd /home/ben/qdrant && source venv/bin/activate && python qdrant_mcp_server.py"
echo ""
echo "测试Qdrant-Ollama MCP服务器..."
echo '{"jsonrpc":"2.0","id":1,"method":"initialize"}' | ssh ben@dev1 "cd /home/ben/qdrant && source venv/bin/activate && ./start_mcp_server.sh"
echo ""
echo "测试完成。"

158
tests/mcp_servers/test_mcp_servers_comprehensive.py
View File

@@ -1,158 +0,0 @@
#!/usr/bin/env python3
"""
测试MCP服务器的脚本
"""
import asyncio
import json
import subprocess
import sys
from typing import Dict, Any, List
async def test_mcp_server(server_name: str, command: List[str], env: Dict[str, str] = None):
"""测试MCP服务器"""
print(f"\n=== 测试 {server_name} 服务器 ===")
# 设置环境变量
process_env = {}
if env:
process_env.update(env)
try:
# 启动服务器进程
process = await asyncio.create_subprocess_exec(
*command,
stdin=asyncio.subprocess.PIPE,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
env=process_env
)
# 初始化请求
init_request = {
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {
"tools": {}
}
}
}
# 发送初始化请求
process.stdin.write((json.dumps(init_request) + "\n").encode())
await process.stdin.drain()
# 读取初始化响应
init_response = await process.stdout.readline()
if init_response:
try:
init_data = json.loads(init_response.decode())
print(f"初始化响应: {init_data}")
except json.JSONDecodeError:
print(f"初始化响应解析失败: {init_response}")
# 获取工具列表
tools_request = {
"jsonrpc": "2.0",
"id": 2,
"method": "tools/list"
}
# 发送工具列表请求
process.stdin.write((json.dumps(tools_request) + "\n").encode())
await process.stdin.drain()
# 读取工具列表响应
tools_response = await process.stdout.readline()
if tools_response:
try:
tools_data = json.loads(tools_response.decode())
print(f"工具列表: {json.dumps(tools_data, indent=2, ensure_ascii=False)}")
# 如果有搜索工具，测试搜索功能
if "result" in tools_data and "tools" in tools_data["result"]:
for tool in tools_data["result"]["tools"]:
tool_name = tool.get("name")
if tool_name and ("search" in tool_name or "document" in tool_name):
print(f"\n测试工具: {tool_name}")
# 测试搜索工具
search_request = {
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
"params": {
"name": tool_name,
"arguments": {
"query": "测试查询",
"limit": 3
}
}
}
# 发送搜索请求
process.stdin.write((json.dumps(search_request) + "\n").encode())
await process.stdin.drain()
# 读取搜索响应
search_response = await process.stdout.readline()
if search_response:
try:
search_data = json.loads(search_response.decode())
print(f"搜索结果: {json.dumps(search_data, indent=2, ensure_ascii=False)}")
except json.JSONDecodeError:
print(f"搜索响应解析失败: {search_response}")
break
except json.JSONDecodeError:
print(f"工具列表响应解析失败: {tools_response}")
# 关闭进程
process.stdin.close()
await process.wait()
except Exception as e:
print(f"测试 {server_name} 服务器时出错: {e}")
async def main():
"""主函数"""
print("开始测试MCP服务器...")
# 测试context7服务器
await test_mcp_server(
"context7",
["npx", "-y", "@upstash/context7-mcp"],
{"DEFAULT_MINIMUM_TOKENS": ""}
)
# 测试qdrant服务器
await test_mcp_server(
"qdrant",
["ssh", "ben@dev1", "cd /home/ben/qdrant && source venv/bin/activate && python qdrant_mcp_server.py"],
{
"QDRANT_URL": "http://dev1:6333",
"QDRANT_API_KEY": "313131",
"COLLECTION_NAME": "mcp",
"EMBEDDING_MODEL": "bge-m3"
}
)
# 测试qdrant-ollama服务器
await test_mcp_server(
"qdrant-ollama",
["ssh", "ben@dev1", "cd /home/ben/qdrant && source venv/bin/activate && ./start_mcp_server.sh"],
{
"QDRANT_URL": "http://dev1:6333",
"QDRANT_API_KEY": "313131",
"COLLECTION_NAME": "ollama_mcp",
"OLLAMA_MODEL": "nomic-embed-text",
"OLLAMA_URL": "http://dev1:11434"
}
)
print("\n所有测试完成。")
if __name__ == "__main__":
asyncio.run(main())

198
tests/mcp_servers/test_mcp_servers_improved.py
View File

@@ -1,198 +0,0 @@
#!/usr/bin/env python3
"""
改进的MCP服务器测试脚本
"""
import asyncio
import json
import subprocess
import sys
from typing import Dict, Any, List, Optional
async def test_mcp_server(server_name: str, command: List[str], env: Dict[str, str] = None):
"""测试MCP服务器"""
print(f"\n=== 测试 {server_name} 服务器 ===")
# 设置环境变量
process_env = {}
if env:
process_env.update(env)
try:
# 启动服务器进程
process = await asyncio.create_subprocess_exec(
*command,
stdin=asyncio.subprocess.PIPE,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
env=process_env
)
# 读取并忽略所有非JSON输出
buffer = ""
while True:
line = await process.stdout.readline()
if not line:
break
line_str = line.decode().strip()
buffer += line_str + "\n"
# 尝试解析JSON
try:
data = json.loads(line_str)
if "jsonrpc" in data:
print(f"收到JSON响应: {json.dumps(data, indent=2, ensure_ascii=False)}")
break
except json.JSONDecodeError:
# 不是JSON，继续读取
continue
# 如果没有找到JSON响应，显示缓冲区内容
if "jsonrpc" not in locals():
print(f"未找到JSON响应，原始输出: {buffer}")
return
# 初始化请求
init_request = {
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {
"tools": {}
}
}
}
# 发送初始化请求
process.stdin.write((json.dumps(init_request) + "\n").encode())
await process.stdin.drain()
# 读取初始化响应
init_response = await read_json_response(process)
if init_response:
print(f"初始化成功")
# 获取工具列表
tools_request = {
"jsonrpc": "2.0",
"id": 2,
"method": "tools/list"
}
# 发送工具列表请求
process.stdin.write((json.dumps(tools_request) + "\n").encode())
await process.stdin.drain()
# 读取工具列表响应
tools_response = await read_json_response(process)
if tools_response:
print(f"工具列表获取成功")
# 如果有搜索工具，测试搜索功能
if "result" in tools_response and "tools" in tools_response["result"]:
for tool in tools_response["result"]["tools"]:
tool_name = tool.get("name")
if tool_name and ("search" in tool_name or "document" in tool_name):
print(f"\n测试工具: {tool_name}")
# 测试搜索工具
search_request = {
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
"params": {
"name": tool_name,
"arguments": {
"query": "测试查询",
"limit": 3
}
}
}
# 发送搜索请求
process.stdin.write((json.dumps(search_request) + "\n").encode())
await process.stdin.drain()
# 读取搜索响应
search_response = await read_json_response(process)
if search_response:
print(f"搜索测试成功")
if "result" in search_response and "content" in search_response["result"]:
for content in search_response["result"]["content"]:
if content.get("type") == "text":
print(f"搜索结果: {content.get('text', '')[:100]}...")
break
# 关闭进程
process.stdin.close()
await process.wait()
except Exception as e:
print(f"测试 {server_name} 服务器时出错: {e}")
async def read_json_response(process):
"""读取JSON响应"""
buffer = ""
while True:
line = await process.stdout.readline()
if not line:
break
line_str = line.decode().strip()
buffer += line_str + "\n"
# 尝试解析JSON
try:
data = json.loads(line_str)
if "jsonrpc" in data:
return data
except json.JSONDecodeError:
# 不是JSON，继续读取
continue
# 如果没有找到JSON响应，返回None
return None
async def main():
"""主函数"""
print("开始测试MCP服务器...")
# 测试context7服务器
await test_mcp_server(
"context7",
["npx", "-y", "@upstash/context7-mcp"],
{"DEFAULT_MINIMUM_TOKENS": ""}
)
# 测试qdrant服务器
await test_mcp_server(
"qdrant",
["ssh", "ben@dev1", "cd /home/ben/qdrant && source venv/bin/activate && python qdrant_mcp_server.py"],
{
"QDRANT_URL": "http://dev1:6333",
"QDRANT_API_KEY": "313131",
"COLLECTION_NAME": "mcp",
"EMBEDDING_MODEL": "bge-m3"
}
)
# 测试qdrant-ollama服务器
await test_mcp_server(
"qdrant-ollama",
["ssh", "ben@dev1", "cd /home/ben/qdrant && source venv/bin/activate && ./start_mcp_server.sh"],
{
"QDRANT_URL": "http://dev1:6333",
"QDRANT_API_KEY": "313131",
"COLLECTION_NAME": "ollama_mcp",
"OLLAMA_MODEL": "nomic-embed-text",
"OLLAMA_URL": "http://dev1:11434"
}
)
print("\n所有测试完成。")
if __name__ == "__main__":
asyncio.run(main())

167
tests/mcp_servers/test_mcp_servers_simple.py
View File

@@ -1,167 +0,0 @@
#!/usr/bin/env python3
"""
简化的MCP服务器测试脚本
"""
import json
import subprocess
import sys
import time
from typing import Dict, Any, List
def test_mcp_server(server_name: str, command: List[str], env: Dict[str, str] = None):
"""测试MCP服务器"""
print(f"\n=== 测试 {server_name} 服务器 ===")
# 设置环境变量
process_env = {}
if env:
process_env.update(env)
try:
# 启动服务器进程
process = subprocess.Popen(
command,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
env=process_env,
text=True
)
# 等待进程启动
time.sleep(2)
# 初始化请求
init_request = {
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {
"tools": {}
}
}
}
# 发送初始化请求
process.stdin.write(json.dumps(init_request) + "\n")
process.stdin.flush()
# 读取初始化响应
init_response = process.stdout.readline()
if init_response:
try:
init_data = json.loads(init_response.strip())
print(f"初始化成功: {init_data.get('result', {}).get('serverInfo', {}).get('name', '未知服务器')}")
except json.JSONDecodeError:
print(f"初始化响应解析失败: {init_response}")
# 获取工具列表
tools_request = {
"jsonrpc": "2.0",
"id": 2,
"method": "tools/list"
}
# 发送工具列表请求
process.stdin.write(json.dumps(tools_request) + "\n")
process.stdin.flush()
# 读取工具列表响应
tools_response = process.stdout.readline()
if tools_response:
try:
tools_data = json.loads(tools_response.strip())
print(f"工具列表获取成功")
# 如果有搜索工具，测试搜索功能
if "result" in tools_data and "tools" in tools_data["result"]:
for tool in tools_data["result"]["tools"]:
tool_name = tool.get("name")
if tool_name and ("search" in tool_name or "document" in tool_name):
print(f"\n测试工具: {tool_name}")
# 测试搜索工具
search_request = {
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
"params": {
"name": tool_name,
"arguments": {
"query": "测试查询",
"limit": 3
}
}
}
# 发送搜索请求
process.stdin.write(json.dumps(search_request) + "\n")
process.stdin.flush()
# 读取搜索响应
search_response = process.stdout.readline()
if search_response:
try:
search_data = json.loads(search_response.strip())
print(f"搜索测试成功")
if "result" in search_data and "content" in search_data["result"]:
for content in search_data["result"]["content"]:
if content.get("type") == "text":
print(f"搜索结果: {content.get('text', '')[:100]}...")
except json.JSONDecodeError:
print(f"搜索响应解析失败: {search_response}")
break
except json.JSONDecodeError:
print(f"工具列表响应解析失败: {tools_response}")
# 关闭进程
process.stdin.close()
process.terminate()
process.wait()
except Exception as e:
print(f"测试 {server_name} 服务器时出错: {e}")
def main():
"""主函数"""
print("开始测试MCP服务器...")
# 测试context7服务器
test_mcp_server(
"context7",
["npx", "-y", "@upstash/context7-mcp"],
{"DEFAULT_MINIMUM_TOKENS": ""}
)
# 测试qdrant服务器
test_mcp_server(
"qdrant",
["ssh", "ben@dev1", "cd /home/ben/qdrant && source venv/bin/activate && python qdrant_mcp_server.py"],
{
"QDRANT_URL": "http://dev1:6333",
"QDRANT_API_KEY": "313131",
"COLLECTION_NAME": "mcp",
"EMBEDDING_MODEL": "bge-m3"
}
)
# 测试qdrant-ollama服务器
test_mcp_server(
"qdrant-ollama",
["ssh", "ben@dev1", "cd /home/ben/qdrant && source venv/bin/activate && ./start_mcp_server.sh"],
{
"QDRANT_URL": "http://dev1:6333",
"QDRANT_API_KEY": "313131",
"COLLECTION_NAME": "ollama_mcp",
"OLLAMA_MODEL": "nomic-embed-text",
"OLLAMA_URL": "http://dev1:11434"
}
)
print("\n所有测试完成。")
if __name__ == "__main__":
main()

189
tests/mcp_servers/test_qdrant_ollama_server.py
View File

@@ -1,189 +0,0 @@
#!/usr/bin/env python3
"""
专门测试qdrant-ollama服务器的脚本
"""
import json
import subprocess
import sys
import time
from typing import Dict, Any, List
def test_qdrant_ollama_server():
"""测试qdrant-ollama服务器"""
print("\n=== 测试 qdrant-ollama 服务器 ===")
try:
# 启动服务器进程
process = subprocess.Popen(
["ssh", "ben@dev1", "cd /home/ben/qdrant && source venv/bin/activate && ./start_mcp_server.sh"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
# 读取并忽略所有非JSON输出
buffer = ""
json_found = False
# 等待进程启动并读取初始输出
for _ in range(10): # 最多尝试10次
line = process.stdout.readline()
if not line:
time.sleep(0.5)
continue
line = line.strip()
buffer += line + "\n"
# 尝试解析JSON
try:
data = json.loads(line)
if "jsonrpc" in data:
json_found = True
print(f"收到JSON响应: {json.dumps(data, indent=2, ensure_ascii=False)}")
break
except json.JSONDecodeError:
# 不是JSON，继续读取
continue
if not json_found:
print(f"未找到JSON响应，原始输出: {buffer}")
process.terminate()
process.wait()
return
# 初始化请求
init_request = {
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {
"tools": {}
}
}
}
# 发送初始化请求
process.stdin.write(json.dumps(init_request) + "\n")
process.stdin.flush()
# 读取初始化响应
init_response = process.stdout.readline()
if init_response:
try:
init_data = json.loads(init_response.strip())
print(f"初始化成功: {init_data.get('result', {}).get('serverInfo', {}).get('name', '未知服务器')}")
except json.JSONDecodeError:
print(f"初始化响应解析失败: {init_response}")
# 获取工具列表
tools_request = {
"jsonrpc": "2.0",
"id": 2,
"method": "tools/list"
}
# 发送工具列表请求
process.stdin.write(json.dumps(tools_request) + "\n")
process.stdin.flush()
# 读取工具列表响应
tools_response = process.stdout.readline()
if tools_response:
try:
tools_data = json.loads(tools_response.strip())
print(f"工具列表获取成功")
# 如果有搜索工具，测试搜索功能
if "result" in tools_data and "tools" in tools_data["result"]:
for tool in tools_data["result"]["tools"]:
tool_name = tool.get("name")
if tool_name and ("search" in tool_name or "document" in tool_name):
print(f"\n测试工具: {tool_name}")
# 先添加一个文档
add_request = {
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
"params": {
"name": "add_document",
"arguments": {
"text": "这是一个测试文档，用于验证qdrant-ollama服务器的功能。",
"metadata": {
"source": "test",
"topic": "测试"
}
}
}
}
# 发送添加文档请求
process.stdin.write(json.dumps(add_request) + "\n")
process.stdin.flush()
# 读取添加文档响应
add_response = process.stdout.readline()
if add_response:
try:
add_data = json.loads(add_response.strip())
print(f"添加文档测试成功")
except json.JSONDecodeError:
print(f"添加文档响应解析失败: {add_response}")
# 测试搜索工具
search_request = {
"jsonrpc": "2.0",
"id": 4,
"method": "tools/call",
"params": {
"name": tool_name,
"arguments": {
"query": "测试文档",
"limit": 3
}
}
}
# 发送搜索请求
process.stdin.write(json.dumps(search_request) + "\n")
process.stdin.flush()
# 读取搜索响应
search_response = process.stdout.readline()
if search_response:
try:
search_data = json.loads(search_response.strip())
print(f"搜索测试成功")
if "result" in search_data and "content" in search_data["result"]:
for content in search_data["result"]["content"]:
if content.get("type") == "text":
print(f"搜索结果: {content.get('text', '')[:100]}...")
except json.JSONDecodeError:
print(f"搜索响应解析失败: {search_response}")
break
except json.JSONDecodeError:
print(f"工具列表响应解析失败: {tools_response}")
# 关闭进程
process.stdin.close()
process.terminate()
process.wait()
except Exception as e:
print(f"测试 qdrant-ollama 服务器时出错: {e}")
def main():
"""主函数"""
print("开始测试qdrant-ollama服务器...")
test_qdrant_ollama_server()
print("\n测试完成。")
if __name__ == "__main__":
main()

15
tests/mcp_servers/test_qdrant_ollama_tools.sh
View File

@@ -1,15 +0,0 @@
#!/bin/bash
echo "测试Qdrant-Ollama MCP服务器的search_documents工具..."
# 测试search_documents工具
ssh ben@dev1 "cd /home/ben/qdrant && source venv/bin/activate && echo '{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"tools/call\",\"params\":{\"name\":\"search_documents\",\"arguments\":{\"query\":\"测试查询\",\"limit\":3}}}' | ./start_mcp_server.sh"
echo ""
echo "测试Qdrant-Ollama MCP服务器的add_document工具..."
# 测试add_document工具
ssh ben@dev1 "cd /home/ben/qdrant && source venv/bin/activate && echo '{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"tools/call\",\"params\":{\"name\":\"add_document\",\"arguments\":{\"text\":\"这是一个测试文档\",\"metadata\":{\"source\":\"test\"}}}}' | ./start_mcp_server.sh"
echo ""
echo "测试完成。"

21
tests/mcp_servers/test_qdrant_ollama_tools_fixed.sh
View File

@@ -1,21 +0,0 @@
#!/bin/bash
echo "测试Qdrant-Ollama MCP服务器的search_documents工具（不带filter参数）..."
# 测试search_documents工具（不带filter参数）
ssh ben@dev1 "cd /home/ben/qdrant && source venv/bin/activate && echo '{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"tools/call\",\"params\":{\"name\":\"search_documents\",\"arguments\":{\"query\":\"测试查询\",\"limit\":3}}}' | ./start_mcp_server.sh"
echo ""
echo "测试Qdrant-Ollama MCP服务器的add_document工具..."
# 测试add_document工具
ssh ben@dev1 "cd /home/ben/qdrant && source venv/bin/activate && echo '{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"tools/call\",\"params\":{\"name\":\"add_document\",\"arguments\":{\"text\":\"这是一个测试文档\",\"metadata\":{\"source\":\"test\"}}}}' | ./start_mcp_server.sh"
echo ""
echo "测试Qdrant-Ollama MCP服务器的list_collections工具..."
# 测试list_collections工具
ssh ben@dev1 "cd /home/ben/qdrant && source venv/bin/activate && echo '{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"tools/call\",\"params\":{\"name\":\"list_collections\",\"arguments\":{}}}' | ./start_mcp_server.sh"
echo ""
echo "测试完成。"

15
tests/mcp_servers/test_search_documents.sh
View File

@@ -1,15 +0,0 @@
#!/bin/bash
echo "测试Qdrant-Ollama MCP服务器的search_documents工具（不带filter参数）..."
# 先添加一个文档
echo "添加测试文档..."
ssh ben@dev1 "cd /home/ben/qdrant && source venv/bin/activate && echo '{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"tools/call\",\"params\":{\"name\":\"add_document\",\"arguments\":{\"text\":\"人工智能是计算机科学的一个分支，致力于创建能够执行通常需要人类智能的任务的系统。\",\"metadata\":{\"source\":\"test\",\"topic\":\"AI\"}}}}' | ./start_mcp_server.sh"
echo ""
echo "搜索文档..."
# 测试search_documents工具（不带filter参数）
ssh ben@dev1 "cd /home/ben/qdrant && source venv/bin/activate && echo '{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"tools/call\",\"params\":{\"name\":\"search_documents\",\"arguments\":{\"query\":\"人工智能\",\"limit\":3}}}' | ./start_mcp_server.sh"
echo ""
echo "测试完成。"

116
tests/run_all_tests.sh
View File

@@ -1,116 +0,0 @@
#!/bin/bash
# MCP服务器测试运行器
# 自动运行所有MCP服务器测试脚本
set -e
# 颜色定义
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
NC='\033[0m' # No Color
# 测试目录
TEST_DIR="/root/mgmt/tests/mcp_servers"
REPORT_FILE="/root/mgmt/tests/test_results_$(date +%Y%m%d_%H%M%S).md"
# 检查测试目录是否存在
if [ ! -d "$TEST_DIR" ]; then
echo -e "${RED}错误: 测试目录 $TEST_DIR 不存在${NC}"
exit 1
fi
# 创建测试报告头部
cat > "$REPORT_FILE" << EOF
# MCP服务器测试报告 - $(date '+%Y-%m-%d %H:%M:%S')
## 测试环境
- 测试时间: $(date '+%Y-%m-%d %H:%M:%S')
- 测试目录: $TEST_DIR
- 测试类型: 自动化批量测试
## 测试结果概览
EOF
echo -e "${YELLOW}开始运行MCP服务器测试套件...${NC}"
echo -e "${YELLOW}测试报告将保存到: $REPORT_FILE${NC}\n"
# 测试计数器
TOTAL_TESTS=0
PASSED_TESTS=0
FAILED_TESTS=0
# 运行Shell脚本测试
echo -e "${YELLOW}运行Shell脚本测试...${NC}"
for test_script in "$TEST_DIR"/*.sh; do
if [ -f "$test_script" ]; then
TEST_NAME=$(basename "$test_script")
echo -e "${YELLOW}运行测试: $TEST_NAME${NC}"
# 运行测试脚本
if bash "$test_script" >> "$REPORT_FILE" 2>&1; then
echo -e "${GREEN}✅ $TEST_NAME 通过${NC}"
echo "- ✅ $TEST_NAME: 通过" >> "$REPORT_FILE"
((PASSED_TESTS++))
else
echo -e "${RED}❌ $TEST_NAME 失败${NC}"
echo "- ❌ $TEST_NAME: 失败" >> "$REPORT_FILE"
((FAILED_TESTS++))
fi
((TOTAL_TESTS++))
echo
fi
done
# 运行Python脚本测试
echo -e "${YELLOW}运行Python脚本测试...${NC}"
for test_script in "$TEST_DIR"/*.py; do
if [ -f "$test_script" ]; then
TEST_NAME=$(basename "$test_script")
echo -e "${YELLOW}运行测试: $TEST_NAME${NC}"
# 运行Python测试
if python3 "$test_script" >> "$REPORT_FILE" 2>&1; then
echo -e "${GREEN}✅ $TEST_NAME 通过${NC}"
echo "- ✅ $TEST_NAME: 通过" >> "$REPORT_FILE"
((PASSED_TESTS++))
else
echo -e "${RED}❌ $TEST_NAME 失败${NC}"
echo "- ❌ $TEST_NAME: 失败" >> "$REPORT_FILE"
((FAILED_TESTS++))
fi
((TOTAL_TESTS++))
echo
fi
done
# 更新测试报告
cat >> "$REPORT_FILE" << EOF
## 测试统计
- 总测试数: $TOTAL_TESTS
- 通过测试: $PASSED_TESTS
- 失败测试: $FAILED_TESTS
- 通过率: $((PASSED_TESTS * 100 / TOTAL_TESTS))%
## 详细测试输出
EOF
# 显示测试结果摘要
echo -e "\n${YELLOW}=== 测试完成 ===${NC}"
echo -e "总测试数: $TOTAL_TESTS"
echo -e "通过测试: ${GREEN}$PASSED_TESTS${NC}"
echo -e "失败测试: ${RED}$FAILED_TESTS${NC}"
echo -e "通过率: $((PASSED_TESTS * 100 / TOTAL_TESTS))%"
echo -e "详细报告: $REPORT_FILE"
# 如果所有测试都通过，返回成功
if [ $FAILED_TESTS -eq 0 ]; then
echo -e "\n${GREEN}所有测试均通过！${NC}"
exit 0
else
echo -e "\n${RED}部分测试失败，请查看详细报告。${NC}"
exit 1
fi