tdagent/tools/get_url_content.py

from collections.abc import Sequence
from typing import Literal, Optional, Dict, Any
import json

import gradio as gr
import requests

from tdagent.constants import HttpContentType

# Define valid HTTP methods
HttpMethod = Literal["GET", "POST", "PUT", "DELETE", "PATCH", "HEAD"]

def make_http_request(
    url: str,
    method: HttpMethod = "GET",
    content_type: str = "",
    body: str = "",
    timeout: float = 30,
    custom_headers: str = ""
) -> tuple[str, str]:
    """Make an HTTP request to a URL with specified method and parameters.

    Args:
        url: The URL to make the request to.
        method: HTTP method to use (GET, POST, PUT, DELETE, PATCH, HEAD).
        content_type: Comma-separated string of content types.
        body: Request body for methods that support it (POST, PUT, PATCH).
        timeout: Request timeout in seconds. Defaults to 30.
        custom_headers: JSON string of additional headers.

    Returns:
        A pair of strings (content, error_message).
    """
    # Initialize headers dictionary
    headers = {}

    # Parse content type
    if content_type:
        headers["Accept"] = content_type

    # Parse custom headers
    if custom_headers:
        try:
            custom_headers_dict = json.loads(custom_headers)
            headers.update(custom_headers_dict)
        except json.JSONDecodeError:
            return "", "Invalid JSON format in custom headers"

    # Prepare request parameters
    request_params: Dict[str, Any] = {
        "url": url,
        "headers": headers,
        "timeout": timeout,
    }

    # Add body for methods that support it
    if method in ["POST", "PUT", "PATCH"] and body:
        request_params["data"] = body

    try:
        response = requests.request(method, **request_params)
    except requests.exceptions.MissingSchema as err:
        return "", str(err)
    except requests.exceptions.RequestException as err:
        return "", str(err)

    try:
        response.raise_for_status()
    except requests.HTTPError as err:
        return "", str(err)

    # For HEAD requests, return headers as content
    if method == "HEAD":
        return str(dict(response.headers)), ""

    return response.text, ""

# Create the Gradio interface
gr_make_http_request = gr.Interface(
    fn=make_http_request,
    inputs=[
        gr.Textbox(label="URL"),
        gr.Dropdown(
            choices=["GET", "POST", "PUT", "DELETE", "PATCH", "HEAD"],
            label="HTTP Method",
            value="GET"
        ),
        gr.Textbox(
            label="Content Type",
            placeholder="text/html,application/json"
        ),
        gr.Textbox(
            label="Request Body (for POST/PUT/PATCH)",
            lines=3,
            placeholder='{"key": "value"}'
        ),
        gr.Number(
            label="Timeout (seconds)",
            value=30,
            minimum=1,
            maximum=300
        ),
        gr.Textbox(
            label="Custom Headers (JSON format)",
            placeholder='{"Authorization": "Bearer token"}'
        )
    ],
    outputs=gr.Text(label="Response"),
    title="Make HTTP Requests",
    description=(
        "Make HTTP requests with different methods and parameters. "
        "Supports GET, POST, PUT, DELETE, PATCH, and HEAD methods. "
        "For POST, PUT, and PATCH requests, you can include a request body. "
        "Custom headers can be added in JSON format. "
        "Be cautious when accessing unknown URLs."
    ),
    examples=[
        ["https://google.com", "GET", "text/html", "", 30, ""],
        ["https://api.example.com/data", "POST", "application/json", '{"key": "value"}', 30, '{"Authorization": "Bearer token"}'],
    ],
)

if __name__ == "__main__":
    gr_make_http_request.launch()