open-swe/agent/tools/http_request.py

from typing import Any

import httpx

from ..utils.url_safety import pinned_url as _pinned_url  # noqa: F401  (kept for tests)
from ..utils.url_safety import request_with_safe_redirects
from ..utils.url_safety import resolve_and_validate as _resolve_and_validate  # noqa: F401


def _blocked_response(url: str, reason: str) -> dict[str, Any]:
    return {
        "success": False,
        "status_code": 0,
        "headers": {},
        "content": f"Request blocked: {reason}",
        "url": url,
    }


async def _request_with_safe_redirects(
    client: httpx.AsyncClient,
    method: str,
    url: str,
    **kwargs: Any,
) -> tuple[httpx.Response | None, dict[str, Any] | None]:
    """Thin wrapper over the shared SSRF-safe redirect loop that shapes a blocked
    hop into the tool's error-response dict."""
    response, blocked = await request_with_safe_redirects(client, method, url, **kwargs)
    if blocked is not None:
        blocked_url, reason = blocked
        return None, _blocked_response(blocked_url, reason)
    return response, None


async def http_request(
    url: str,
    method: str = "GET",
    headers: dict[str, str] | None = None,
    data: str | dict | None = None,
    params: dict[str, str] | None = None,
    timeout: int = 30,
) -> dict[str, Any]:
    """Make HTTP requests to APIs and web services.

    Do not use this tool for GitHub API calls. Use `GH_TOKEN=dummy gh` in the
    sandbox so GitHub authentication is handled by the sandbox proxy.

    Args:
        url: Target URL
        method: HTTP method (GET, POST, PUT, DELETE, etc.)
        headers: HTTP headers to include
        data: Request body data (string or dict)
        params: URL query parameters
        timeout: Request timeout in seconds

    Returns:
        Dictionary with response data including status, headers, and content
    """
    try:
        kwargs: dict[str, Any] = {}

        if headers:
            kwargs["headers"] = headers
        if params:
            kwargs["params"] = params
        if data:
            if isinstance(data, dict):
                kwargs["json"] = data
            else:
                kwargs["content"] = data

        async with httpx.AsyncClient(timeout=timeout) as client:
            response, blocked = await _request_with_safe_redirects(
                client,
                method,
                url,
                **kwargs,
            )
        if blocked:
            return blocked

        try:
            content = response.json()
        except ValueError:
            content = response.text

        return {
            "success": response.status_code < 400,
            "status_code": response.status_code,
            "headers": dict(response.headers),
            "content": content,
            "url": str(response.url),
        }

    except httpx.TimeoutException:
        return {
            "success": False,
            "status_code": 0,
            "headers": {},
            "content": f"Request timed out after {timeout} seconds",
            "url": url,
        }
    except httpx.HTTPError as e:
        return {
            "success": False,
            "status_code": 0,
            "headers": {},
            "content": f"Request error: {e!s}",
            "url": url,
        }