Coverage for merco/tools/web_tools.py: 38%

1"""网络工具 - 搜索与抓取""" 

2 

3import httpx 

4from .base import BaseTool 

5 

6 

7class WebFetch(BaseTool): 

8 """抓取网页内容""" 

9 

10 name = "web_fetch" 

11 description = "获取指定 URL 的内容" 

12 toolset = "web" 

13 parameters = { 

14 "type": "object", 

15 "properties": { 

16 "url": {"type": "string", "description": "目标 URL"}, 

17 "format": {"type": "string", "enum": ["text", "html"], "description": "返回格式"}, 

18 }, 

19 "required": ["url"], 

20 } 

21 

22 async def execute(self, url: str, format: str = "text") -> dict: 

23 try: 

24 async with httpx.AsyncClient() as client: 

25 response = await client.get(url, follow_redirects=True, timeout=30) 

26 response.raise_for_status() 

27 

28 if format == "html": 

29 return {"content": response.text, "url": url} 

30 else: 

31 # 简单提取文本内容 

32 import re 

33 text = re.sub(r"<[^>]+>", "", response.text) 

34 text = re.sub(r"\s+", " ", text).strip() 

35 return {"content": text, "url": url} 

36 

37 except Exception as e: 

38 return {"error": str(e)} 

39 

40 

41class WebSearch(BaseTool): 

42 """网络搜索 — DuckDuckGo (免费，无需 API key)""" 

43 

44 name = "web_search" 

45 description = "搜索网络获取信息。返回标题、URL 和摘要。" 

46 toolset = "web" 

47 parameters = { 

48 "type": "object", 

49 "properties": { 

50 "query": {"type": "string", "description": "搜索关键词"}, 

51 "n": {"type": "integer", "description": "结果数量，默认 5"}, 

52 }, 

53 "required": ["query"], 

54 } 

55 

56 async def execute(self, query: str, n: int = 5) -> dict: 

57 try: 

58 from ddgs import DDGS 

59 results = [] 

60 with DDGS() as ddgs: 

61 for r in ddgs.text(query, max_results=n): 

62 results.append({ 

63 "title": r["title"], 

64 "url": r["href"], 

65 "snippet": r["body"], 

66 }) 

67 return {"results": results, "query": query} 

68 except ImportError: 

69 return {"error": "ddgs 未安装。运行 pip install ddgs", "results": []} 

70 except Exception as e: 

71 return {"error": str(e), "results": []} 

72