Execute tool via stream_manager.
(self, tool_name: str, arguments: dict[str, Any])
| 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.""" |
| 112 | if hasattr(self.stream_manager, "call_tool"): |
| 113 | return await self.stream_manager.call_tool(tool_name, arguments) |
| 114 | else: |
| 115 | raise ValueError("Stream manager doesn't support tool execution") |
| 116 | |
| 117 | async def get_server_for_tool(self, tool_name: str) -> str: |
| 118 | """Get server for tool from stream_manager.""" |
no test coverage detected