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

1""" 

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

3 

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

5""" 

6 

7from __future__ import annotations 

8 

9import asyncio 

10import json 

11from collections.abc import Callable 

12 

13from agentos.llm.base import Tool as LLMTool 

14from agentos.llm.base import ToolFunction, ToolParameter 

15from agentos.tools.base import BaseTool 

16from agentos.tools.registry import ToolRegistry 

17 

18 

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

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

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

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

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

24 

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

26 tool_params[name] = ToolParameter( 

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

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

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

30 required=name in required_list, 

31 ) 

32 

33 return LLMTool( 

34 function=ToolFunction( 

35 name=tool.name, 

36 description=tool.description, 

37 parameters=tool_params, 

38 required=required_list, 

39 ) 

40 ) 

41 

42 

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

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

45 

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

47 try: 

48 loop = asyncio.get_event_loop() 

49 if loop.is_running(): 

50 import concurrent.futures 

51 

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

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

54 result = future.result(timeout=30) 

55 else: 

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

57 except RuntimeError: 

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

59 

60 if result.error: 

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

62 return result.output or "" 

63 

64 return sync_handler 

65 

66 

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

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

69 for name in registry.list_names(): 

70 tool = registry.get(name) 

71 if tool is None: 

72 continue 

73 llm_tool = base_tool_to_llm_tool(tool) 

74 handler = make_handler(tool) 

75 executor.register(llm_tool, handler)