Coverage for agentos/tools/http_tools.py: 21%

1"""HTTP 工具 — HTTP 请求、文件下载。""" 

2 

3from __future__ import annotations 

4 

5import json 

6import os 

7import tempfile 

8import time 

9from urllib.parse import urlparse 

10 

11from agentos.tools.base import BaseTool, ToolResult 

12 

13 

14class HttpRequestTool(BaseTool): 

15 """HTTP 请求工具 — 发送 GET/POST/PUT/DELETE 请求。""" 

16 

17 name = "http_request" 

18 description = "发送 HTTP 请求（GET/POST/PUT/DELETE），支持 JSON body、自定义 header" 

19 

20 @property 

21 def parameters(self) -> dict: 

22 return { 

23 "type": "object", 

24 "properties": { 

25 "url": {"type": "string", "description": "请求 URL"}, 

26 "method": { 

27 "type": "string", 

28 "description": "HTTP 方法：GET/POST/PUT/DELETE，默认 GET", 

29 "enum": ["GET", "POST", "PUT", "DELETE", "PATCH"], 

30 }, 

31 "body": {"type": "string", "description": "请求体（JSON 字符串）"}, 

32 "headers": {"type": "string", "description": "自定义 Header，JSON 格式串"}, 

33 "timeout": {"type": "integer", "description": "超时秒数，默认 30"}, 

34 }, 

35 "required": ["url"], 

36 } 

37 

38 async def execute(self, arguments: dict, sandbox=None) -> ToolResult: 

39 import urllib.error 

40 import urllib.request 

41 

42 url = arguments.get("url", "") 

43 method = arguments.get("method", "GET").upper() 

44 body = arguments.get("body", "") 

45 headers_str = arguments.get("headers", "{}") 

46 timeout = arguments.get("timeout", 30) 

47 

48 try: 

49 parsed_headers = json.loads(headers_str) if headers_str else {} 

50 except json.JSONDecodeError: 

51 return ToolResult.fail(call_id="", error=f"Invalid headers JSON: {headers_str}") 

52 

53 data = body.encode("utf-8") if body else None 

54 req = urllib.request.Request(url, data=data, method=method) 

55 req.add_header("User-Agent", "AgentOS-HttpTool/1.0") 

56 req.add_header("Accept", "application/json, text/plain, */*") 

57 if body: 

58 req.add_header("Content-Type", "application/json") 

59 for k, v in parsed_headers.items(): 

60 req.add_header(k, str(v)) 

61 

62 t0 = time.time() 

63 try: 

64 with urllib.request.urlopen(req, timeout=timeout) as resp: 

65 elapsed_ms = (time.time() - t0) * 1000 

66 raw_body = resp.read() 

67 text_body = raw_body.decode("utf-8", errors="replace") 

68 content_type = resp.headers.get("Content-Type", "") 

69 

70 output = ( 

71 f"Status: {resp.status}\n" 

72 f"Content-Type: {content_type}\n" 

73 f"Body length: {len(raw_body)} bytes\n" 

74 f"Elapsed: {elapsed_ms:.0f}ms\n\n" 

75 f"{text_body[:3000]}" 

76 ) 

77 return ToolResult.ok(call_id="", output=output) 

78 

79 except urllib.error.HTTPError as e: 

80 elapsed_ms = (time.time() - t0) * 1000 

81 error_body = "" 

82 try: 

83 error_body = e.read().decode("utf-8", errors="replace")[:1000] 

84 except Exception: 

85 pass 

86 return ToolResult.ok( 

87 call_id="", 

88 output=f"HTTP {e.code} {e.reason}\nElapsed: {elapsed_ms:.0f}ms\n\n{error_body}", 

89 ) 

90 except Exception as e: 

91 return ToolResult.fail(call_id="", error=f"Request failed: {e}") 

92 

93 

94class DownloadTool(BaseTool): 

95 """文件下载工具 — 下载 URL 内容到本地文件。""" 

96 

97 name = "download_file" 

98 description = "从 URL 下载文件到本地，返回本地路径和文件大小" 

99 

100 @property 

101 def parameters(self) -> dict: 

102 return { 

103 "type": "object", 

104 "properties": { 

105 "url": {"type": "string", "description": "下载 URL"}, 

106 "output_path": { 

107 "type": "string", 

108 "description": "输出目录或文件路径，默认临时目录", 

109 }, 

110 }, 

111 "required": ["url"], 

112 } 

113 

114 async def execute(self, arguments: dict, sandbox=None) -> ToolResult: 

115 import urllib.request 

116 

117 url = arguments.get("url", "") 

118 output_path = arguments.get("output_path", "") 

119 

120 parsed = urlparse(url) 

121 filename = os.path.basename(parsed.path) or "download" 

122 if output_path: 

123 if os.path.isdir(output_path) or output_path.endswith("/"): 

124 filepath = os.path.join(output_path, filename) 

125 else: 

126 filepath = output_path 

127 else: 

128 filepath = os.path.join(tempfile.gettempdir(), filename) 

129 

130 # Avoid overwriting 

131 if os.path.exists(filepath): 

132 base, ext = os.path.splitext(filename) 

133 counter = 1 

134 while os.path.exists(filepath): 

135 filepath = os.path.join(os.path.dirname(filepath), f"{base}_{counter}{ext}") 

136 counter += 1 

137 

138 os.makedirs(os.path.dirname(filepath) or ".", exist_ok=True) 

139 

140 t0 = time.time() 

141 try: 

142 with urllib.request.urlopen(url, timeout=300) as resp: 

143 total = 0 

144 with open(filepath, "wb") as f: 

145 while True: 

146 chunk = resp.read(8192) 

147 if not chunk: 

148 break 

149 f.write(chunk) 

150 total += len(chunk) 

151 

152 elapsed_ms = (time.time() - t0) * 1000 

153 size_mb = total / (1024 * 1024) 

154 return ToolResult.ok( 

155 call_id="", 

156 output=f"Downloaded: {filepath}\nSize: {total} bytes ({size_mb:.2f} MB)\nTime: {elapsed_ms:.0f}ms", 

157 ) 

158 except Exception as e: 

159 return ToolResult.fail(call_id="", error=f"Download failed: {e}")