needle / live_api_audit.py
Leon4gr45's picture
Upload folder using huggingface_hub
faaafef verified
Raw History Blame Contribute Delete
2.39 kB
import httpx
import json
import time
URL = "https://leon4gr45-needle.hf.space"
# Optional mode shouldn't strictly require this for basic tests, but let's see
# The API requires a valid key for privileged mutation in optional mode. But for read it's fine.
def run():
client = httpx.Client(timeout=30.0)
traces = []
def trace_req(method, path, payload=None, headers=None):
t0 = time.time()
try:
if payload:
res = client.request(method, f"{URL}{path}", json=payload, headers=headers)
else:
res = client.request(method, f"{URL}{path}", headers=headers)
t1 = time.time()
traces.append({
"method": method,
"path": path,
"payload": payload,
"status": res.status_code,
"response": res.text[:500] if len(res.text) > 500 else res.json() if "{" in res.text else res.text,
"latency_ms": round((t1 - t0) * 1000, 2),
"timestamp": t0
})
print(f"{method} {path} -> {res.status_code}")
except Exception as e:
print(f"Error {method} {path}: {e}")
# 1. Health
trace_req("GET", "/health")
# 2. Ready
trace_req("GET", "/ready")
# 3. Info
trace_req("GET", "/v1/info")
# 4. Tool registry retrieval
trace_req("GET", "/v1/tools")
# 5. Tool candidate retrieval
trace_req("POST", "/v1/tools/retrieve", payload={"query": "test query", "max_candidates": 5})
# 6. /v1/route
trace_req("POST", "/v1/route", payload={"input": "test", "tools": [{"name": "test", "description": "desc", "parameters": {"type": "object"}}]})
# 7. Hierarchical gate routing
trace_req("POST", "/v1/gates/route", payload={"query": "test", "tools": [{"name": "test", "description": "desc", "parameters": {"type": "object"}}], "max_candidates": 2})
# 8. OpenAI compatible endpoint
trace_req("POST", "/v1/chat/completions", payload={"model": "needle3", "messages": [{"role": "user", "content": "test"}], "tools": [{"name": "test", "description": "desc", "parameters": {"type": "object"}}]})
# 9. Metrics
trace_req("GET", "/v1/metrics")
with open("tests/live/evidence/api-traces.jsonl", "w") as f:
for t in traces:
f.write(json.dumps(t) + "\n")
if __name__ == '__main__':
run()