File size: 3,962 Bytes
53b2ac9 | 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 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 | # SPDX-License-Identifier: MIT
"""
HTTP Request Plugin for TrashClaw
Make HTTP requests with custom headers, methods, and body.
Supports GET, POST, PUT, DELETE, PATCH methods.
Returns response status, headers, and body.
Plugin contract:
TOOL_DEF = dict with name, description, parameters (OpenAI function schema)
run(**kwargs) -> str (called with the parsed arguments)
"""
import urllib.request
import urllib.error
import json
from typing import Optional, Dict, Any
TOOL_DEF = {
"name": "http_request",
"description": "Make HTTP requests with custom headers, methods, and body. Returns response status, headers, and body (JSON parsed if applicable).",
"parameters": {
"type": "object",
"properties": {
"url": {
"type": "string",
"description": "The URL to request (e.g., 'https://api.example.com/data')"
},
"method": {
"type": "string",
"enum": ["GET", "POST", "PUT", "DELETE", "PATCH"],
"description": "HTTP method (default: GET)"
},
"headers": {
"type": "object",
"description": "Custom headers as key-value pairs (e.g., {'Authorization': 'Bearer token'})",
"additionalProperties": {"type": "string"}
},
"body": {
"type": "string",
"description": "Request body (will be sent as JSON if Content-Type is application/json)"
},
"timeout": {
"type": "integer",
"description": "Request timeout in seconds (default: 30)"
}
},
"required": ["url"]
}
}
def run(url: str, method: str = "GET", headers: Optional[Dict[str, str]] = None,
body: Optional[str] = None, timeout: int = 30, **kwargs) -> str:
"""
Make an HTTP request and return formatted response.
Args:
url: The URL to request
method: HTTP method (GET, POST, PUT, DELETE, PATCH)
headers: Custom headers dict
body: Request body string
timeout: Request timeout in seconds
Returns:
Formatted response string with status, headers, and body
"""
# Prepare headers
req_headers = {"User-Agent": "TrashClaw-HTTP/1.0"}
if headers:
req_headers.update(headers)
# Prepare body
data = None
if body:
data = body.encode("utf-8")
if "Content-Type" not in req_headers:
req_headers["Content-Type"] = "application/json"
# Create request
req = urllib.request.Request(url, data=data, headers=req_headers, method=method.upper())
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
status = resp.status
resp_headers = dict(resp.headers)
raw_body = resp.read().decode("utf-8", errors="replace")
# Try to parse JSON
try:
parsed_body = json.loads(raw_body)
body_str = json.dumps(parsed_body, indent=2, ensure_ascii=False)
except (json.JSONDecodeError, ValueError):
body_str = raw_body[:2000] # Limit response size
if len(raw_body) > 2000:
body_str += f"\n... (truncated, {len(raw_body)} total chars)"
# Format output
result = f"HTTP {method.upper()} {url}\n"
result += f"Status: {status}\n"
result += f"Headers:\n"
for k, v in list(resp_headers.items())[:10]: # Limit headers shown
result += f" {k}: {v}\n"
result += f"\nBody:\n{body_str}"
return result
except urllib.error.HTTPError as e:
return f"HTTP Error {e.code}: {e.reason}\nURL: {url}"
except urllib.error.URLError as e:
return f"URL Error: {e.reason}\nURL: {url}"
except Exception as e:
return f"Request failed: {type(e).__name__}: {e}"
|