Get the server name for a given tool.
(self, tool_name: str)
| 1232 | return [] |
| 1233 | |
| 1234 | async def get_server_for_tool(self, tool_name: str) -> str | None: |
| 1235 | """Get the server name for a given tool.""" |
| 1236 | if not self.stream_manager: |
| 1237 | return None |
| 1238 | |
| 1239 | try: |
| 1240 | tools = await self.get_all_tools() |
| 1241 | tool = next((t for t in tools if t.name == tool_name), None) |
| 1242 | return tool.namespace if tool else None |
| 1243 | |
| 1244 | except Exception as e: |
| 1245 | logger.error(f"Error getting server for tool: {e}") |
| 1246 | return None |
| 1247 | |
| 1248 | def get_streams(self): |
| 1249 | """Get active streams from StreamManager.""" |