Files

93 lines
3.1 KiB
Python

#!/usr/bin/env python3
"""Test di concorrenza MCP: N chiamate parallele + sonde /health e ping.
Uso (dentro il container):
docker exec loogle-mcp python /srv/scripts/test_mcp_concurrency.py [N] [tool]
Misura se una chiamata lenta blocca le altre (event loop bloccato) e la
latenza delle ricerche semantiche.
"""
import concurrent.futures as cf
import json
import os
import sys
import time
import httpx
sys.path.insert(0, "/srv")
from app.db import get_conn # noqa: E402
from app.jwt_utils import create_access_token # noqa: E402
BASE = os.environ.get("TEST_BASE", "http://127.0.0.1:8700")
N = int(sys.argv[1]) if len(sys.argv) > 1 else 4
TOOL = sys.argv[2] if len(sys.argv) > 2 else "search_knowledge"
CLIENT_TIMEOUT = 150.0
row = get_conn().execute("SELECT id, username FROM users WHERE username='daniele'").fetchone()
token, _ = create_access_token(
{"id": row["id"], "username": row["username"]},
"context:read knowledge:read gitea:read admin",
"loadtest",
)
HEADERS = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
def rpc(method: str, params: dict, req_id: int) -> tuple[float, str]:
t0 = time.monotonic()
try:
r = httpx.post(
f"{BASE}/mcp",
headers=HEADERS,
json={"jsonrpc": "2.0", "id": req_id, "method": method, "params": params},
timeout=CLIENT_TIMEOUT,
)
body = r.json()
if "error" in body:
status = f"ERR {body['error'].get('code')}: {body['error'].get('message', '')[:90]}"
else:
text = body["result"]["content"][0]["text"] if method == "tools/call" else "{}"
payload = json.loads(text)
n = len(payload.get("results", [])) if isinstance(payload, dict) else "?"
status = f"OK results={n}"
except Exception as exc: # noqa: BLE001
status = f"EXC {type(exc).__name__}: {exc}"[:100]
return time.monotonic() - t0, status
def health() -> float:
t0 = time.monotonic()
try:
httpx.get(f"{BASE}/health", timeout=CLIENT_TIMEOUT)
except Exception: # noqa: BLE001
return -1.0
return time.monotonic() - t0
def main() -> None:
queries = ["failover keepalived", "irrigazione zone", "turni medici", "backup nas",
"runbook", "rete vlan", "home assistant", "paperless"]
t_start = time.monotonic()
with cf.ThreadPoolExecutor(max_workers=N + 2) as pool:
futs = {
pool.submit(rpc, "tools/call",
{"name": TOOL, "arguments": {"query": queries[i % len(queries)], "limit": 3}},
i + 1): f"{TOOL}#{i + 1}"
for i in range(N)
}
time.sleep(0.5)
h = pool.submit(health)
p = pool.submit(rpc, "tools/call", {"name": "ping", "arguments": {}}, 999)
print(f"/health durante il carico: {h.result():.2f}s")
pt, ps = p.result()
print(f"ping durante il carico: {pt:.2f}s {ps}")
for fut in cf.as_completed(futs):
dt, status = fut.result()
print(f"{futs[fut]:<24} {dt:6.2f}s {status}")
print(f"Totale: {time.monotonic() - t_start:.2f}s (N={N})")
if __name__ == "__main__":
main()