| 90 | return '\n\n'.join(texts) |
| 91 | |
| 92 | async def get_tools(self) -> Dict: |
| 93 | tools = {} |
| 94 | for key, session in self.sessions.items(): |
| 95 | tools[key] = [] |
| 96 | try: |
| 97 | response = await session.list_tools() |
| 98 | except Exception as e: |
| 99 | new_eg = enhance_error( |
| 100 | e, f'MCP `{key}` list tool failed, details: ') |
| 101 | raise new_eg from e |
| 102 | _session_tools = response.tools |
| 103 | exclude = [] |
| 104 | include = [] |
| 105 | if self.include_functions: |
| 106 | if key in self.include_functions: |
| 107 | include = self.include_functions[key] |
| 108 | elif self.exclude_functions: |
| 109 | if key in self.exclude_functions: |
| 110 | exclude = self.exclude_functions[key] |
| 111 | _session_tools = [ |
| 112 | t for t in _session_tools if t.name not in exclude |
| 113 | ] |
| 114 | if include: |
| 115 | _session_tools = [ |
| 116 | t for t in _session_tools if t.name in include |
| 117 | ] |
| 118 | _session_tools = [ |
| 119 | Tool( |
| 120 | tool_name=t.name, |
| 121 | server_name=key, |
| 122 | description=t.description, |
| 123 | parameters=t.inputSchema) for t in _session_tools |
| 124 | ] |
| 125 | tools[key].extend(_session_tools) |
| 126 | return tools |
| 127 | |
| 128 | @staticmethod |
| 129 | def print_tools(server_name: str, tools: ListToolsResult): |