50 lines
1.7 KiB
Python
50 lines
1.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Regressione: chiama ogni tool MCP senza argomenti obbligatori (sola lettura) e ne misura la durata.
|
|
|
|
Uso (dentro il container):
|
|
docker exec loogle-mcp python /srv/scripts/test_mcp_readonly_tools.py
|
|
"""
|
|
|
|
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")
|
|
# Tool senza argomenti obbligatori ma con effetti collaterali: esclusi
|
|
SKIP = {"reindex_apps"}
|
|
|
|
row = get_conn().execute("SELECT id, username FROM users WHERE username='daniele'").fetchone()
|
|
scope = ("context:read knowledge:read gitea:read home:read irrigation:read turni:read admin")
|
|
token, _ = create_access_token({"id": row["id"], "username": row["username"]}, scope, "regression")
|
|
HEADERS = {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
def rpc(method: str, params: dict) -> dict:
|
|
r = httpx.post(f"{BASE}/mcp", headers=HEADERS, timeout=120,
|
|
json={"jsonrpc": "2.0", "id": 1, "method": method, "params": params})
|
|
return r.json()
|
|
|
|
|
|
tools = rpc("tools/list", {})["result"]["tools"]
|
|
failures = 0
|
|
for tool in tools:
|
|
if tool["inputSchema"].get("required") or tool["name"] in SKIP:
|
|
continue
|
|
t0 = time.monotonic()
|
|
body = rpc("tools/call", {"name": tool["name"], "arguments": {}})
|
|
dt = time.monotonic() - t0
|
|
if "error" in body:
|
|
status = f"ERR {body['error']['message'][:80]}"
|
|
failures += 1
|
|
else:
|
|
status = f"OK {len(body['result']['content'][0]['text'])} bytes"
|
|
print(f"{tool['name']:<30} {dt:6.2f}s {status}")
|
|
print(f"Tool totali: {len(tools)}, errori: {failures}")
|