Spaces:
Running
Running
File size: 3,765 Bytes
473bd03 24480a0 473bd03 24480a0 473bd03 24480a0 473bd03 24480a0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 | """
backend/tools/trigger_webhook.py — P17-F4: Tool trigger_webhook generico.
Invia una HTTP request (GET/POST/PUT/PATCH/DELETE) a qualsiasi URL esterno.
Importato da registry.py tramite: from tools.trigger_webhook import trigger_webhook
Allowlist: WEBHOOK_ALLOWED_HOSTS env (vuoto = tutti consentiti).
"""
from __future__ import annotations
import json
import logging
import os
from typing import Any
import httpx
_logger = logging.getLogger("agente_ai.tools.trigger_webhook")
_ALLOWED_METHODS = {"GET", "POST", "PUT", "PATCH", "DELETE"}
_ALLOWED_HOSTS_RAW = os.getenv("WEBHOOK_ALLOWED_HOSTS", "")
_ALLOWED_HOSTS: set[str] = {h.strip().lower() for h in _ALLOWED_HOSTS_RAW.split(",") if h.strip()}
def _check_host(url: str) -> tuple[bool, str]:
if not _ALLOWED_HOSTS:
return True, ""
try:
from urllib.parse import urlparse
host = urlparse(url).hostname or ""
if host.lower() in _ALLOWED_HOSTS:
return True, ""
for allowed in _ALLOWED_HOSTS:
if allowed.startswith("*.") and host.lower().endswith(allowed[1:]):
return True, ""
return False, f"Host '{host}' non nella allowlist WEBHOOK_ALLOWED_HOSTS"
except Exception as exc:
return False, f"URL non valido: {exc}"
async def trigger_webhook(
url: str,
payload: "dict[str, Any] | str | None" = None,
method: str = "POST",
headers: "dict[str, str] | None" = None,
timeout: float = 10.0,
) -> dict:
"""
Invia una HTTP request a un URL esterno.
Args:
url: URL destinazione (obbligatorio).
payload: Body JSON (dict) o stringa raw.
method: GET|POST|PUT|PATCH|DELETE (default POST).
headers: Header HTTP aggiuntivi.
timeout: Max 15s.
Returns:
{ok, status_code, body, url, method} | {ok: False, error: str}
"""
method = method.upper().strip()
if method not in _ALLOWED_METHODS:
return {"ok": False, "error": f"Metodo non supportato: {method}. Usa: {sorted(_ALLOWED_METHODS)}"}
ok_host, err_host = _check_host(url)
if not ok_host:
return {"ok": False, "error": err_host}
_hdrs: dict[str, str] = {"User-Agent": "AgentAI-Webhook/1.0"}
if headers:
_hdrs.update(headers)
body_bytes: bytes | None = None
if payload is not None:
if isinstance(payload, dict):
body_bytes = json.dumps(payload).encode("utf-8")
_hdrs.setdefault("Content-Type", "application/json")
else:
body_bytes = str(payload).encode("utf-8")
_hdrs.setdefault("Content-Type", "text/plain; charset=utf-8")
_timeout = min(float(timeout), 15.0)
try:
async with httpx.AsyncClient(timeout=_timeout, follow_redirects=True) as client:
resp = await client.request(
method, url, headers=_hdrs,
content=body_bytes if method != "GET" else None,
params=payload if method == "GET" and isinstance(payload, dict) else None,
)
resp_body: Any
ct = resp.headers.get("content-type", "")
try:
resp_body = resp.json() if "application/json" in ct else resp.text[:4000]
except Exception:
resp_body = resp.text[:4000]
_logger.info("[trigger_webhook] %s %s -> HTTP %d", method, url, resp.status_code)
return {"ok": resp.is_success, "status_code": resp.status_code, "body": resp_body, "url": url, "method": method, "error": None}
except httpx.TimeoutException:
return {"ok": False, "error": f"Timeout {_timeout}s — URL non raggiungibile: {url}"}
except httpx.ConnectError as exc:
return {"ok": False, "error": f"Connessione fallita a {url}: {exc}"}
except Exception as exc:
_logger.warning("[trigger_webhook] errore: %s", exc)
return {"ok": False, "error": str(exc)[:300]}
|