add
测试mcp基座
This commit is contained in:
1 parent
c0966ad334
commit
1afcc10a61
6 files changed
+489
-96
No files matched your search
+32
-53
@@ -1,64 +1,43 @@
|
||||
import socket
|
||||
import json
|
||||
import time
|
||||
import base64
|
||||
|
||||
def send_request(req_data, silent=False):
|
||||
try:
|
||||
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
client.connect(("127.0.0.1", 8080))
|
||||
|
||||
# 发送 UTF-8 编码的 JSON 请求
|
||||
client.sendall(json.dumps(req_data).encode('utf-8'))
|
||||
|
||||
# 接收并解码结果
|
||||
response = client.recv(4096)
|
||||
|
||||
if not silent:
|
||||
print(">>> 收到回复:")
|
||||
print(json.dumps(json.loads(response.decode('utf-8')), indent=2, ensure_ascii=False))
|
||||
print("-" * 50)
|
||||
|
||||
client.close()
|
||||
except Exception as e:
|
||||
print(f"连接失败: {e}")
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("=== 测试 1: 验证 C++ DLL 工具调用 ===")
|
||||
request_call = {
|
||||
def test_screenshot():
|
||||
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
client.connect(("127.0.0.1", 8080))
|
||||
|
||||
req = {
|
||||
"jsonrpc": "2.0",
|
||||
"id": 4,
|
||||
"id": 5,
|
||||
"method": "tools/call",
|
||||
"params": {
|
||||
"name": "draw_circle_cpp",
|
||||
"arguments": {
|
||||
"cx": 2000.0,
|
||||
"cy": 2000.0,
|
||||
"r": 500.0
|
||||
}
|
||||
"name": "get_viewport_screenshot",
|
||||
"arguments": {}
|
||||
}
|
||||
}
|
||||
send_request(request_call)
|
||||
client.sendall(json.dumps(req).encode('utf-8'))
|
||||
|
||||
# 稍作停顿,方便观察 CAD 屏幕
|
||||
time.sleep(1)
|
||||
# 图片 base64 报文可能很大,需要循环接收直到拿完
|
||||
response_data = b""
|
||||
while True:
|
||||
chunk = client.recv(8192)
|
||||
response_data += chunk
|
||||
if len(chunk) < 8192:
|
||||
break
|
||||
|
||||
client.close()
|
||||
|
||||
print("\n=== 测试 2: C++ 并发写入压力测试 (生成10个同心圆) ===")
|
||||
for i in range(10):
|
||||
radius = 100.0 + i * 50.0
|
||||
req = {
|
||||
"jsonrpc": "2.0",
|
||||
"id": 100 + i,
|
||||
"method": "tools/call",
|
||||
"params": {
|
||||
"name": "draw_circle_cpp",
|
||||
"arguments": {
|
||||
"cx": 4000.0,
|
||||
"cy": 2000.0,
|
||||
"r": radius
|
||||
}
|
||||
}
|
||||
}
|
||||
# 连续快速发送,不打印详细返回以模拟极限并发
|
||||
send_request(req, silent=True)
|
||||
resp_json = json.loads(response_data.decode('utf-8'))
|
||||
|
||||
print(">>> 并发指令发送完毕,请检查 CAD 屏幕。")
|
||||
# 解析并保存图片
|
||||
content_list = resp_json["result"]["content"]
|
||||
for item in content_list:
|
||||
if item["type"] == "image":
|
||||
base64_data = item["data"]
|
||||
image_bytes = base64.b64decode(base64_data)
|
||||
with open("test_screenshot.png", "wb") as f:
|
||||
f.write(image_bytes)
|
||||
print("截图已保存为 test_screenshot.png,请在当前目录查看!")
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_screenshot()
|
||||
Reference in new issue
Block a user