66 lines
3.2 KiB
Python
66 lines
3.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Conformità Streamable HTTP / lifecycle MCP: negoziazione versione, 202 per notifiche, header.
|
|
|
|
Uso (dentro il container):
|
|
docker exec loogle-mcp python /srv/scripts/test_mcp_protocol.py
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
|
|
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") + "/mcp"
|
|
row = get_conn().execute("SELECT id, username FROM users WHERE username='daniele'").fetchone()
|
|
token, _ = create_access_token(dict(row), "context:read", "protocol-test")
|
|
AUTH = {"Authorization": f"Bearer {token}"}
|
|
failures = 0
|
|
|
|
|
|
def check(label: str, cond: bool, detail: str = "") -> None:
|
|
global failures
|
|
failures += 0 if cond else 1
|
|
print(f"{'OK ' if cond else 'FAIL'} {label} {detail}")
|
|
|
|
|
|
def init(version):
|
|
params = {"protocolVersion": version, "capabilities": {}, "clientInfo": {"name": "protocol-test"}}
|
|
r = httpx.post(BASE, json={"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": params})
|
|
return r.json()["result"]["protocolVersion"]
|
|
|
|
|
|
for requested, expected in [("2025-06-18", "2025-06-18"), ("2025-03-26", "2025-03-26"),
|
|
("2024-11-05", "2024-11-05"), ("2099-01-01", "2025-06-18")]:
|
|
got = init(requested)
|
|
check(f"initialize {requested}", got == expected, f"-> {got}")
|
|
|
|
r = httpx.post(BASE, json={"jsonrpc": "2.0", "method": "notifications/initialized"})
|
|
check("notifica non autenticata initialized -> 202", r.status_code == 202 and not r.content, str(r.status_code))
|
|
r = httpx.post(BASE, headers=AUTH, json={"jsonrpc": "2.0", "method": "notifications/cancelled",
|
|
"params": {"requestId": 5}})
|
|
check("notifica cancelled -> 202", r.status_code == 202, str(r.status_code))
|
|
r = httpx.post(BASE, json={"jsonrpc": "2.0", "method": "notifications/cancelled"})
|
|
check("notifica non autenticata non ammessa -> 401", r.status_code == 401, str(r.status_code))
|
|
r = httpx.post(BASE, headers=AUTH, json=[{"jsonrpc": "2.0", "method": "notifications/initialized"},
|
|
{"jsonrpc": "2.0", "id": 7, "method": "ping"}])
|
|
body = r.json()
|
|
check("batch notifica+request -> solo 1 risposta", r.status_code == 200 and len(body) == 1 and body[0]["id"] == 7,
|
|
str(body))
|
|
r = httpx.post(BASE, headers={**AUTH, "MCP-Protocol-Version": "2025-06-18"},
|
|
json={"jsonrpc": "2.0", "id": 2, "method": "tools/list"})
|
|
check("header versione valida -> 200", r.status_code == 200 and "tools" in r.json()["result"], str(r.status_code))
|
|
r = httpx.post(BASE, headers={**AUTH, "MCP-Protocol-Version": "1999-01-01"},
|
|
json={"jsonrpc": "2.0", "id": 3, "method": "tools/list"})
|
|
check("header versione non supportata -> 400", r.status_code == 400, str(r.status_code))
|
|
r = httpx.post(BASE, json={"jsonrpc": "2.0", "id": 4, "method": "tools/list"})
|
|
check("request senza token -> 401 + WWW-Authenticate",
|
|
r.status_code == 401 and "resource_metadata" in r.headers.get("www-authenticate", ""), str(r.status_code))
|
|
r = httpx.post(BASE, headers=AUTH, json=[])
|
|
check("batch vuoto -> 400", r.status_code == 400, str(r.status_code))
|
|
print(f"Errori: {failures}")
|
|
sys.exit(1 if failures else 0)
|