Coverage for agentos/tools/bridge.py: 78%

1 

2""" 

3Tool Bridge — 连接 ToolRegistry (BaseTool) 和 ToolExecutor (Tool + callable)。 

4 

5让 BaseTool 子类可以无缝注册到 ToolAgent 使用的 ToolExecutor 中。 

6""" 

7from __future__ import annotations 

8import json, asyncio 

9from typing import Callable 

10from agentos.tools.base import BaseTool 

11from agentos.tools.registry import ToolRegistry 

12from agentos.llm.base import Tool as LLMTool, ToolFunction, ToolParameter 

13 

14 

15def base_tool_to_llm_tool(tool: BaseTool) -> LLMTool: 

16 """将 BaseTool 的 parameters schema 转换为 LLM Tool 对象。""" 

17 params = tool.parameters or {"type": "object", "properties": {}, "required": []} 

18 tool_params: dict[str, ToolParameter] = {} 

19 required_list: list[str] = params.get("required", []) 

20 

21 for name, schema in params.get("properties", {}).items(): 

22 tool_params[name] = ToolParameter( 

23 type=schema.get("type", "string"), 

24 description=schema.get("description", ""), 

25 enum=schema.get("enum"), 

26 required=name in required_list, 

27 ) 

28 

29 return LLMTool( 

30 function=ToolFunction( 

31 name=tool.name, 

32 description=tool.description, 

33 parameters=tool_params, 

34 required=required_list, 

35 ) 

36 ) 

37 

38 

39def make_handler(tool: BaseTool) -> Callable[..., str]: 

40 """创建适配 callable，让 ToolExecutor 能调用 BaseTool。""" 

41 def sync_handler(**kwargs) -> str: 

42 try: 

43 loop = asyncio.get_event_loop() 

44 if loop.is_running(): 

45 import concurrent.futures 

46 with concurrent.futures.ThreadPoolExecutor() as executor: 

47 future = executor.submit(asyncio.run, tool.execute(kwargs)) 

48 result = future.result(timeout=30) 

49 else: 

50 result = asyncio.run(tool.execute(kwargs)) 

51 except RuntimeError: 

52 result = asyncio.run(tool.execute(kwargs)) 

53 

54 if result.error: 

55 return json.dumps({"error": result.error}) 

56 return result.output or "" 

57 

58 return sync_handler 

59 

60 

61def bridge_registry_to_executor(registry: ToolRegistry, executor) -> None: 

62 """将 ToolRegistry 中所有已注册的 BaseTool 桥接到 ToolExecutor。""" 

63 for name in registry.list_names(): 

64 tool = registry.get(name) 

65 if tool is None: 

66 continue 

67 llm_tool = base_tool_to_llm_tool(tool) 

68 handler = make_handler(tool) 

69 executor.register(llm_tool, handler)