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
}