"""Universal HTTP request tool supporting any method, headers, cookies, and data."""
import httpx
import json
import logging
from typing import Optional
from tools._safe_http import assert_safe_http_url, safe_http_headers
from tools.alter_privileges import PRIVILEGES, has_privilege
logger = logging.getLogger(__name__)
VALID_METHODS = {"GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"}
TOOL_NAME = "http_request"
TOOL_DESCRIPTION = (
"Make an HTTP request with any method, headers, cookies, and body "
"to any URL. Returns status code, headers, body, and timing."
)
TOOL_PARAMETERS = {
"type": "object",
"properties": {
"url": {
"type": "string",
"description": "The URL to send the request to.",
},
"method": {
"type": "string",
"enum": ["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"],
"description": "HTTP method (default: GET).",
},
"headers": {
"type": "string",
"description": (
'JSON string of headers, e.g. \'{"Authorization": "Bearer token"}\'.'
),
},
"cookies": {
"type": "string",
"description": 'JSON string of cookies, e.g. \'{"session_id": "abc123"}\'.',
},
"data": {
"type": "string",
"description": "Raw body content as string.",
},
"json_data": {
"type": "string",
"description": (
"JSON string to send as request body. "
"Automatically sets Content-Type to application/json."
),
},
"form_data": {
"type": "string",
"description": (
"JSON string of form fields for "
"application/x-www-form-urlencoded encoding."
),
},
"params": {
"type": "string",
"description": 'JSON string of URL query parameters, e.g. \'{"page": "1"}\'.',
},
"timeout": {
"type": "number",
"description": "Request timeout in seconds (default: 30).",
},
"verify_ssl": {
"type": "boolean",
"description": "Verify SSL certificates (default: true).",
},
"follow_redirects": {
"type": "boolean",
"description": "Follow HTTP redirects (default: true).",
},
},
"required": ["url"],
}
def _parse_json_param(name: str, value: Optional[str], expect_dict: bool = True):
"""Parse a JSON-string parameter. Returns (parsed, error_json_or_None)."""
if not value:
return ({} if expect_dict else None), None
try:
parsed = json.loads(value)
if expect_dict and not isinstance(parsed, dict):
return None, json.dumps({
"error": f"{name} must be a JSON object",
"details": f"Got {type(parsed).__name__} instead of dict",
})
return parsed, None
except json.JSONDecodeError as e:
return None, json.dumps({
"error": f"Invalid JSON in {name} parameter",
"details": str(e),
})
[docs]
async def run(
url: str,
method: str = "GET",
headers: Optional[str] = None,
cookies: Optional[str] = None,
data: Optional[str] = None,
json_data: Optional[str] = None,
form_data: Optional[str] = None,
params: Optional[str] = None,
timeout: float = 30.0,
verify_ssl: bool = True,
follow_redirects: bool = True,
ctx=None,
) -> str:
"""Execute this tool and return the result.
Args:
url (str): URL string.
method (str): The method value.
headers (Optional[str]): The headers value.
cookies (Optional[str]): The cookies value.
data (Optional[str]): Input data payload.
json_data (Optional[str]): The json data value.
form_data (Optional[str]): The form data value.
params (Optional[str]): The params value.
timeout (float): Maximum wait time in seconds.
verify_ssl (bool): The verify ssl value.
follow_redirects (bool): The follow redirects value.
ctx: Tool execution context providing access to bot internals.
Returns:
str: Result string.
"""
method = method.upper().strip()
if method not in VALID_METHODS:
return json.dumps({
"error": f"Invalid HTTP method: {method}",
"details": f"Valid methods are: {', '.join(sorted(VALID_METHODS))}",
})
if not url or not url.strip():
return json.dumps({"error": "URL is required and cannot be empty"})
url = url.strip()
if not url.startswith(("http://", "https://")):
return json.dumps({
"error": "Invalid URL format",
"details": "URL must start with http:// or https://",
})
try:
url = assert_safe_http_url(url)
except ValueError as exc:
return json.dumps({"error": "URL not allowed", "details": str(exc)})
if not verify_ssl:
allow_tls_off = False
if ctx is not None:
redis = getattr(ctx, "redis", None)
cfg = getattr(ctx, "config", None)
uid = getattr(ctx, "user_id", None)
if redis is not None and cfg is not None and uid is not None:
allow_tls_off = await has_privilege(
redis, uid, PRIVILEGES["UNSANDBOXED_EXEC"], cfg,
)
if not allow_tls_off:
verify_ssl = True
parsed_headers, err = _parse_json_param("headers", headers)
if err:
return err
parsed_cookies, err = _parse_json_param("cookies", cookies)
if err:
return err
parsed_params, err = _parse_json_param("params", params)
if err:
return err
parsed_json_data = None
if json_data:
try:
parsed_json_data = json.loads(json_data)
except json.JSONDecodeError as e:
return json.dumps({
"error": "Invalid JSON in json_data parameter",
"details": str(e),
})
parsed_form_data, err = _parse_json_param("form_data", form_data)
if err:
return err
if not form_data:
parsed_form_data = None
try:
timeout = float(timeout)
if timeout <= 0:
timeout = 30.0
elif timeout > 300:
timeout = 300.0
except (TypeError, ValueError):
timeout = 30.0
request_kwargs = {
"method": method,
"url": url,
"timeout": timeout,
"follow_redirects": follow_redirects,
}
if parsed_headers:
request_kwargs["headers"] = safe_http_headers(parsed_headers)
if parsed_cookies:
request_kwargs["cookies"] = parsed_cookies
if parsed_params:
request_kwargs["params"] = parsed_params
body_count = sum([
data is not None,
parsed_json_data is not None,
parsed_form_data is not None,
])
if body_count > 1:
return json.dumps({
"error": "Multiple body types specified",
"details": "Use only one of: data, json_data, or form_data",
})
if parsed_json_data is not None:
request_kwargs["json"] = parsed_json_data
elif parsed_form_data is not None:
request_kwargs["data"] = parsed_form_data
elif data is not None:
request_kwargs["content"] = data
try:
logger.info(f"HTTP {method} request to {url}")
async with httpx.AsyncClient(verify=verify_ssl) as client:
response = await client.request(**request_kwargs)
response_headers = dict(response.headers)
elapsed_ms = response.elapsed.total_seconds() * 1000
try:
body = response.text
except Exception:
body = "<binary content>"
result = {
"status_code": response.status_code,
"headers": response_headers,
"body": body,
"url": str(response.url),
"elapsed_ms": round(elapsed_ms, 2),
}
logger.info(f"HTTP {method} {url} completed: {response.status_code} in {elapsed_ms:.2f}ms")
return json.dumps(result, ensure_ascii=False)
except httpx.TimeoutException:
return json.dumps({
"error": "Request timed out",
"details": f"No response received within {timeout} seconds",
})
except httpx.ConnectError as e:
return json.dumps({"error": "Connection failed", "details": str(e)})
except httpx.TooManyRedirects:
return json.dumps({
"error": "Too many redirects",
"details": "The request exceeded the maximum number of redirects",
})
except httpx.RequestError as e:
return json.dumps({
"error": f"Request error: {type(e).__name__}",
"details": str(e),
})
except Exception as e:
logger.error(f"Unexpected error in http_request: {e}", exc_info=True)
return json.dumps({
"error": f"Unexpected error: {type(e).__name__}",
"details": str(e),
})