| |
| """ |
| 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 |
| """ |
| |
| req_headers = {"User-Agent": "TrashClaw-HTTP/1.0"} |
| if headers: |
| req_headers.update(headers) |
|
|
| |
| data = None |
| if body: |
| data = body.encode("utf-8") |
| if "Content-Type" not in req_headers: |
| req_headers["Content-Type"] = "application/json" |
|
|
| |
| 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: |
| 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] |
| if len(raw_body) > 2000: |
| body_str += f"\n... (truncated, {len(raw_body)} total chars)" |
|
|
| |
| 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]: |
| 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}" |
|
|
