File size: 3,481 Bytes
5374a2d |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 |
from typing import Dict, Any, Optional, List
from .tool import Tool, Toolkit
from .request_base import RequestBase
class HTTPRequestTool(Tool):
"""Universal HTTP request tool that handles all request methods and processing."""
name: str = "http_request"
description: str = "Make HTTP requests (GET, POST, PUT, DELETE, etc.) with automatic content processing and optional file saving"
inputs: Dict[str, Dict[str, str]] = {
"url": {
"type": "string",
"description": "The URL to make the request to"
},
"method": {
"type": "string",
"description": "HTTP method to use (GET, POST, PUT, DELETE, PATCH, etc.). Defaults to GET"
},
"headers": {
"type": "object",
"description": "Optional headers to include in the request"
},
"params": {
"type": "object",
"description": "Optional URL parameters to include in the request"
},
"data": {
"type": "object",
"description": "Optional form data to send in the request body"
},
"json_data": {
"type": "object",
"description": "Optional JSON data to send in the request body"
},
"return_raw": {
"type": "boolean",
"description": "If true, return raw response content. If false (default), return processed content (HTML converted to text, JSON parsed, etc.)"
},
"save_file_path": {
"type": "string",
"description": "Optional file path to save the response content"
}
}
required: Optional[List[str]] = ["url"]
def __init__(self, request_base: RequestBase = None):
super().__init__()
self.request_base = request_base
def __call__(self, url: str, method: str = 'GET', headers: dict = None,
params: dict = None, data: dict = None,
json_data: dict = None, return_raw: bool = False,
save_file_path: str = None) -> Dict[str, Any]:
"""
Make an HTTP request with comprehensive processing and error handling.
Args:
url: The URL to request
method: HTTP method (GET, POST, PUT, DELETE, etc.)
headers: Optional headers to include
params: Optional URL parameters
data: Optional form data to send
json_data: Optional JSON data to send
return_raw: If True, return raw content; if False, return processed content
save_file_path: Optional path to save the response content
Returns:
Dictionary containing response data and metadata
"""
return self.request_base.request_and_process(
url=url,
method=method,
headers=headers,
params=params,
data=data,
json_data=json_data,
return_raw=return_raw,
save_file_path=save_file_path
)
class RequestToolkit(Toolkit):
def __init__(self, name: str = "RequestToolkit"):
# Create the shared request base instance
request_base = RequestBase()
# Create tools with the shared request base
tools = [
HTTPRequestTool(request_base=request_base)
]
# Initialize parent with tools
super().__init__(name=name, tools=tools)
|