Test-specific tool initialization.
(self, on_progress=None)
| 75 | return cls(stream_manager, model_manager) |
| 76 | |
| 77 | async def _initialize_tools(self, on_progress=None) -> None: |
| 78 | """Test-specific tool initialization.""" |
| 79 | # Get tools from stream_manager |
| 80 | if hasattr(self.stream_manager, "get_internal_tools"): |
| 81 | self.tools = list(self.stream_manager.get_internal_tools()) |
| 82 | else: |
| 83 | self.tools = list(self.stream_manager.get_all_tools()) |
| 84 | |
| 85 | # Get server info |
| 86 | self.server_info = list(self.stream_manager.get_server_info()) |
| 87 | |
| 88 | # Build mappings - tools are ToolInfo objects |
| 89 | self.tool_to_server_map = { |
| 90 | t.name: self.stream_manager.get_server_for_tool(t.name) for t in self.tools |
| 91 | } |
| 92 | |
| 93 | # Convert tools to OpenAI format for tests |
| 94 | self.openai_tools = [ |
| 95 | { |
| 96 | "type": "function", |
| 97 | "function": { |
| 98 | "name": t.name, |
| 99 | "description": t.description or "", |
| 100 | "parameters": t.parameters or {}, |
| 101 | }, |
| 102 | } |
| 103 | for t in self.tools |
| 104 | ] |
| 105 | self.tool_name_mapping = {} |
| 106 | |
| 107 | # Copy for system prompt |
| 108 | self.internal_tools = list(self.tools) |
| 109 | |
| 110 | async def execute_tool(self, tool_name: str, arguments: dict[str, Any]) -> Any: |
| 111 | """Execute tool via stream_manager.""" |
no test coverage detected