Source code for aidputils.agents.tools.http_tool

"""
HTTP Tool for executing HTTPS requests to external APIs.

This tool provides a standardized interface for agents to make HTTP requests
with template variable substitution, SSRF protection, and MCP-compliant output formatting.
"""

import logging
from typing import Any, Dict, Optional

import requests
from oci import retry as oci_retry

from aidputils.agents.tools.base_tool import BaseTool
from aidputils.agents.auth.client.generic_rest_client import GenericRestClient
from aidputils.agents.tools.http.template_engine import (
    substitute_url,
    substitute_headers,
    substitute_body,
    substitute_template,
    TemplateError,
    SecurityError,
)
from aidputils.agents.tools.http.url_validator import (
    validate_url,
    SSRFError,
)
from aidputils.agents.tools.http.http_auth import build_auth_for_http
from aidputils.agents.tools.http.response_optimizer import optimize_response
from aidputils.agents.tools.http.error_handler import (
    ErrorCode,
    classify_error,
    build_error_response,
    format_error_for_mcp,
)
from aidputils.agents.tools.http.audit_logger import AuditContext
from aidputils.agents.toolkit.service_metrics_util import get_service_metrics

logger = logging.getLogger(__name__)


def _json_safe(value: Any) -> Any:
    """Convert common non-JSON-serializable types to JSON-safe equivalents."""
    if value is None or isinstance(value, (str, int, float, bool)):
        return value

    if isinstance(value, bytes):
        return value.decode("utf-8", errors="replace")

    if isinstance(value, dict):
        return {str(k): _json_safe(v) for k, v in value.items()}

    if isinstance(value, (list, tuple)):
        return [_json_safe(v) for v in value]

    return str(value)

# Metrics: HTTP tool (module-level cached references; lazily initialized)
_service_metrics = None
_requests_counter = None
_errors_counter = None
_duration_hist = None


def _reset_http_metrics() -> None:
    global _service_metrics, _requests_counter, _errors_counter, _duration_hist
    _service_metrics = None
    _requests_counter = None
    _errors_counter = None
    _duration_hist = None


def _init_http_metrics() -> None:
    """Initialize HTTP tool metrics.

    Intentionally lazy (called from `_emit_request_metrics`) rather than
    import-time so we don't miss metrics when OpenTelemetry is configured later
    in process startup.
    """

    global _service_metrics, _requests_counter, _errors_counter, _duration_hist

    if _service_metrics is not None:
        return

    try:
        _service_metrics = get_service_metrics()
        _requests_counter = _service_metrics.create_counter(
            "http_tool.requests.total", description="Total HTTP tool requests"
        )
        _errors_counter = _service_metrics.create_counter(
            "http_tool.errors.total", description="Total HTTP tool errors"
        )
        _duration_hist = _service_metrics.create_histogram(
            "http_tool.request.duration_ms",
            description="HTTP tool request duration in milliseconds",
        )
    except Exception:
        logger.exception("Failed to initialize HTTP tool metrics")
        _reset_http_metrics()

# MCP output keys
CONTENT_KEY = "content"
STRUCTURED_CONTENT_KEY = "structuredContent"
IS_ERROR_KEY = "isError"

# Supported HTTP methods
SUPPORTED_METHODS = {"GET", "POST", "PUT", "DELETE", "PATCH"}

# Header size limits
MAX_HEADER_SIZE = 8 * 1024  # 8KB per header
MAX_TOTAL_HEADERS_SIZE = 32 * 1024  # 32KB total

# Sensitive headers that cannot be set via templating
SENSITIVE_HEADERS = {
    "authorization",
    "cookie",
    "x-api-key",
    "x-auth-token",
}

# Retry configuration
DEFAULT_TIMEOUT = 30  # seconds
MAX_RETRY_ATTEMPTS = 5
DEFAULT_RETRY_ATTEMPTS = 3

# Response size limit (10MB per design spec)
MAX_RESPONSE_SIZE = 10 * 1024 * 1024  # 10MB


def _emit_request_metrics(
    method: str,
    status_code: Optional[int],
    auth_type: str,
    elapsed_ms: float,
    is_error: bool,
    error_type: Optional[str] = None
) -> None:
    """
    Emit HTTP tool metrics.
    
    Metrics emitted:
    - http_tool.requests.total: Counter with method, status_code, auth_type
    - http_tool.errors.total: Counter with method, error_type
    - http_tool.request.duration_ms: Histogram with method, status_code
    """
    try:
        if _service_metrics is None or _requests_counter is None or _duration_hist is None:
            _init_http_metrics()

        if _service_metrics is None:
            return
        
        # Common attributes
        base_attrs = {"method": method, "auth_type": auth_type}
        
        # Request counter (always emit)
        if _requests_counter:
            attrs = {**base_attrs, "status_code": str(status_code) if status_code else "error"}
            _service_metrics.increment_counter(_requests_counter, 1, attributes=attrs)
        
        # Error counter (emit on errors)
        if is_error and _errors_counter:
            attrs = {"method": method, "error_type": error_type or "unknown"}
            _service_metrics.increment_counter(_errors_counter, 1, attributes=attrs)
        
        # Duration histogram (always emit)
        if _duration_hist:
            attrs = {"method": method, "status_code": str(status_code) if status_code else "error"}
            _service_metrics.record_histogram(_duration_hist, elapsed_ms, attributes=attrs)
            
    except Exception:
        logger.exception("Failed to emit HTTP tool metrics")


[docs] def validate_headers(headers: Dict[str, str]) -> None: """ Validate headers for size limits and sensitive header protection. Args: headers: Dictionary of header name -> value Raises: SecurityError: If headers violate security constraints """ if not headers: return total_size = 0 for name, value in headers.items(): # Check for sensitive headers if name.lower() in SENSITIVE_HEADERS: raise SecurityError( f"Header '{name}' is sensitive and cannot be set directly. " "Use the auth.authType configuration instead." ) # Check for sensitive header prefixes if name.lower().startswith("x-auth-"): raise SecurityError( f"Header '{name}' matches sensitive pattern 'X-Auth-*' and cannot be set directly." ) # Sanitize header value (strip newlines to prevent header injection) if "\r" in value or "\n" in value: raise SecurityError( f"Header '{name}' contains newline characters which are not allowed" ) # Check individual header size header_size = len(name) + len(value) + 4 # +4 for ": " and "\r\n" if header_size > MAX_HEADER_SIZE: raise SecurityError( f"Header '{name}' exceeds maximum size of {MAX_HEADER_SIZE} bytes" ) total_size += header_size # Check total headers size if total_size > MAX_TOTAL_HEADERS_SIZE: raise SecurityError( f"Total headers size ({total_size} bytes) exceeds maximum of {MAX_TOTAL_HEADERS_SIZE} bytes" )
[docs] @BaseTool.register class HttpTool(BaseTool): """ HTTP Tool for executing HTTPS requests to external APIs. Supports GET, POST, PUT, DELETE, PATCH methods with: - {{variable}} template substitution in URL, headers, params, and body - SSRF protection (blocks private IPs, cloud metadata endpoints) - Header size limits and sensitive header protection - JSON body serialization - MCP-compliant output formatting Configuration: name: Tool name for agent discovery description: Tool description for LLM tool selection method: HTTP method (GET, POST, PUT, DELETE, PATCH) url: Target URL with optional {{variable}} templates headers: Optional custom headers (dict) params: Optional query parameters (dict) body: Optional request body (dict or JSON-serializable) timeout: Optional timeout in seconds (default: 30) auth: Optional auth configuration object: {"authType": "NO_AUTH"|"BEARER_TOKEN"|"OAUTH", ...} retry: Optional retry configuration {"max_attempts": 1-5} (default: 3) Example: conf = { "name": "get_user", "description": "Get user details by ID", "method": "GET", "url": "https://api.example.com/users/{{user_id}}", "headers": {"X-Custom-Header": "{{custom_value}}"}, "auth": {"authType": "NO_AUTH"} } runtime_params = {"user_id": "123", "custom_value": "test"} result = HttpTool.invoke(conf, runtime_params) """ @classmethod def _invoke_tool(cls, conf: Dict[str, Any], runtime_params: Dict[str, Any], **context_vars) -> Dict[str, Any]: """ Execute an HTTP request based on configuration and runtime parameters. Args: conf: Tool configuration containing method, url, headers, params, body runtime_params: Runtime values for template substitution **context_vars: Additional context (unused in M1) Returns: Dict containing response data or error information """ runtime_params = runtime_params or {} # Extract and validate method method = conf.get("method", "GET").upper() if method not in SUPPORTED_METHODS: return build_error_response( ErrorCode.INVALID_METHOD, f"Unsupported HTTP method: {method}. Supported: {', '.join(SUPPORTED_METHODS)}" ) # Extract URL (required) url_template = conf.get("url") if not url_template: return build_error_response( ErrorCode.MISSING_URL, "URL is required in tool configuration" ) # Initialize auth variables before templating/auth try-block so request # exception handlers can always emit metrics without UnboundLocalError. auth_type = "NO_AUTH" signer = None auth_headers = None try: # Substitute URL with security validation and URL encoding url = substitute_url(url_template, runtime_params) # Validate URL for SSRF protection validate_url(url, resolve_dns=True) # Substitute headers with security validation headers = substitute_headers(conf.get("headers", {}), runtime_params) # Validate headers for size limits and sensitive headers validate_headers(headers) # Substitute params params = substitute_template( conf.get("params", {}), runtime_params, url_encode=False, validate_security=True ) # Substitute body (more permissive) body = substitute_body(conf.get("body"), runtime_params) if conf.get("body") else None # Build auth (breaking change): use auth strategies via conf['auth'] auth_headers, signer, auth_type = build_auth_for_http(conf.get("auth")) if auth_headers: # Auth headers are intentionally merged after validate_headers(), since # templating disallows sensitive headers like Authorization. headers = {**(headers or {}), **auth_headers} except TemplateError as e: return build_error_response( ErrorCode.TEMPLATE_ERROR, str(e) ) except SecurityError as e: return build_error_response( ErrorCode.SECURITY_ERROR, str(e) ) except SSRFError as e: return build_error_response( ErrorCode.SSRF_BLOCKED, str(e) ) except ValueError as e: return build_error_response( ErrorCode.AUTH_ERROR, str(e) ) # Get timeout (default 30s) timeout = conf.get("timeout", DEFAULT_TIMEOUT) # Get retry configuration retry_config = conf.get("retry", {}) retry_attempts = retry_config.get("max_attempts", DEFAULT_RETRY_ATTEMPTS) # Cap retry attempts at MAX_RETRY_ATTEMPTS retry_attempts = min(max(1, retry_attempts), MAX_RETRY_ATTEMPTS) # Create retry strategy if retry is enabled retry_strategy = None if retry_attempts > 1: retry_strategy = oci_retry.DEFAULT_RETRY_STRATEGY # Create REST client with signer for authentication client = GenericRestClient( endpoint="", # Full URL provided in path_or_url signer=signer, timeout=timeout, default_headers=headers if headers else None, retry_strategy=retry_strategy ) # Execute request with audit logging with AuditContext(method, url, headers) as audit: try: # Prepare request kwargs request_kwargs = { "path_or_url": url, "params": params if params else None, } # Add body for methods that support it if method in {"POST", "PUT", "PATCH"} and body is not None: request_kwargs["json_body"] = body # Execute request using GenericRestClient response = client.request(method, **request_kwargs) # Get status code for audit and MCP output status_code = getattr(response, "status", None) if status_code is None: status_code = getattr(response, "status_code", 200) audit.set_response(status_code) # Get response data and headers for optimization response_data = getattr(response, "data", response) response_headers = dict(getattr(response, "headers", {})) # Check response size limit content_length = response_headers.get("content-length", response_headers.get("Content-Length")) if content_length: try: content_length_value = int(str(content_length).strip()) except (ValueError, TypeError): content_length_value = None if content_length_value is not None and content_length_value > MAX_RESPONSE_SIZE: elapsed_ms = audit.elapsed_ms or 0 return build_error_response( ErrorCode.RESPONSE_TOO_LARGE, f"Response size ({content_length_value} bytes) exceeds maximum allowed ({MAX_RESPONSE_SIZE} bytes)", url=url, method=method, elapsed_seconds=elapsed_ms / 1000, ) content_type = response_headers.get("content-type", response_headers.get("Content-Type", "")) # Apply response optimization if configured optimization_config = conf.get("responseOptimization", {}) if optimization_config: response_data = optimize_response(response_data, content_type, optimization_config) # Emit success metrics _emit_request_metrics( method=method, status_code=status_code, auth_type=auth_type, elapsed_ms=audit.elapsed_ms or 0, is_error=False ) # Return response data for MCP formatting return { "response": response, "response_data": response_data, "response_headers": response_headers, "status_code": status_code, "url": url, "method": method, "optimized": bool(optimization_config) } except requests.Timeout as e: elapsed_ms = audit.elapsed_ms or 0 elapsed_s = elapsed_ms / 1000 logger.warning(f"HTTP request timed out after {elapsed_s:.2f}s: {method} {url}") # Emit error metrics _emit_request_metrics( method=method, status_code=None, auth_type=auth_type, elapsed_ms=elapsed_ms, is_error=True, error_type="TIMEOUT" ) return build_error_response( ErrorCode.TIMEOUT, f"Request timed out after {elapsed_s:.2f}s (configured timeout: {timeout}s)", url=url, method=method, elapsed_seconds=elapsed_s, ) except Exception as e: elapsed_ms = audit.elapsed_ms or 0 elapsed_s = elapsed_ms / 1000 logger.exception(f"HTTP request failed after {elapsed_s:.2f}s: {method} {url}") error_code = classify_error(e) # Emit error metrics _emit_request_metrics( method=method, status_code=None, auth_type=auth_type, elapsed_ms=elapsed_ms, is_error=True, error_type=error_code.value if hasattr(error_code, 'value') else str(error_code) ) return build_error_response( error_code, str(e), url=url, method=method, elapsed_seconds=elapsed_s, ) @classmethod def _format_output_as_mcp(cls, result: Dict[str, Any]) -> Dict[str, Any]: """ Format the response as MCP-compliant output. MCP Output Format: { "content": [{"type": "text"|"json", "text"|"json": ...}], "structuredContent": {"statusCode": int, "headers": dict, "body": any}, "isError": bool } Args: result: Result dict from _invoke_tool Returns: MCP-compliant response dict """ # Handle errors from _invoke_tool (uses structured error_handler format) if result.get("error"): return format_error_for_mcp(result) response = result.get("response") # Handle missing response if response is None: return { CONTENT_KEY: [ { "type": "text", "text": "Tool execution failed: no response returned" } ], IS_ERROR_KEY: True } # Extract response data from GenericRestClient response try: # Use pre-extracted/optimized data if available response_data = result.get("response_data") response_headers = result.get("response_headers", {}) # Fallback to extracting from response object if response_data is None: response = result.get("response") response_data = getattr(response, "data", response) response_headers = dict(getattr(response, "headers", {})) safe_headers = _json_safe(response_headers) if not isinstance(safe_headers, dict): safe_headers = {} # Determine content type content_type = safe_headers.get("content-type", safe_headers.get("Content-Type", "")) # Build content array # Use a simple text representation to avoid double-escaping when LangGraph serializes contents = [] if "application/json" in str(content_type).lower() or isinstance(response_data, (dict, list)): # JSON response - provide brief summary, full data is in structuredContent if isinstance(response_data, list): text_content = f"JSON array with {len(response_data)} items" elif isinstance(response_data, dict): keys = list(response_data.keys())[:5] text_content = f"JSON object with keys: {', '.join(keys)}" if len(response_data.keys()) > 5: text_content += f" (and {len(response_data.keys()) - 5} more)" else: text_content = str(response_data) contents.append({ "type": "json", "text": text_content }) else: # Text or binary response text_content = _json_safe(response_data) if not isinstance(text_content, str): text_content = str(text_content) contents.append({ "type": "text", "text": text_content }) # Build structured content (including status code per design spec) status_code = result.get("status_code") is_error = isinstance(status_code, int) and status_code >= 400 safe_body = _json_safe(response_data) structured_content = { "statusCode": status_code, "headers": safe_headers, "body": safe_body, } return { CONTENT_KEY: contents, STRUCTURED_CONTENT_KEY: structured_content, IS_ERROR_KEY: is_error } except Exception as e: logger.exception("Error formatting response") return { CONTENT_KEY: [ { "type": "text", "text": f"Error formatting response: {str(e)}" } ], IS_ERROR_KEY: True }