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

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 Any, Callable 

10from agentos.tools.base import BaseTool, ToolCall, ToolResult 

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)