Coverage for agentos/tools/web_tools.py: 46%

1""" 

2网络工具 — 搜索与网页抓取。 

3""" 

4 

5from __future__ import annotations 

6 

7import httpx 

8 

9from agentos.tools.base import BaseTool, PermissionLevel, ToolResult 

10 

11 

12class WebFetchTool(BaseTool): 

13 """网页抓取工具。""" 

14 

15 name = "web_fetch" 

16 description = "抓取指定URL的网页正文内容。用于读取网页、文档、API响应等。" 

17 permission_level = PermissionLevel.SAFE 

18 

19 @property 

20 def parameters(self) -> dict: 

21 return { 

22 "type": "object", 

23 "properties": { 

24 "url": { 

25 "type": "string", 

26 "description": "要抓取的URL，需以http://或https://开头", 

27 }, 

28 }, 

29 "required": ["url"], 

30 } 

31 

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

33 url = arguments["url"] 

34 try: 

35 async with httpx.AsyncClient(timeout=30, follow_redirects=True) as client: 

36 response = await client.get( 

37 url, 

38 headers={"User-Agent": "AgentOS/0.1 (+https://agentos.dev)"}, 

39 ) 

40 response.raise_for_status() 

41 text = response.text[:10000] # 限制最大10K字符 

42 return ToolResult.ok("", output=text) 

43 except httpx.HTTPStatusError as e: 

44 return ToolResult.fail("", error=f"HTTP {e.response.status_code}: {url}") 

45 except httpx.TimeoutException: 

46 return ToolResult.fail("", error=f"Timeout fetching {url}") 

47 except Exception as e: 

48 return ToolResult.fail("", error=str(e))