Query each registered MCP server and return formatted tool signatures.
(self)
| 58 | ) |
| 59 | |
| 60 | async def get_server_descriptions(self) -> dict[str, str]: |
| 61 | """Query each registered MCP server and return formatted tool signatures.""" |
| 62 | descriptions: dict[str, str] = {} |
| 63 | for name, path in self._server_paths.items(): |
| 64 | try: |
| 65 | tools = await _list_tools(path) |
| 66 | lines = [] |
| 67 | for t in tools: |
| 68 | params = ", ".join( |
| 69 | f"{p['name']}: {p['type']}{'?' if not p['required'] else ''}" |
| 70 | for p in t.get("parameters", []) |
| 71 | ) |
| 72 | lines.append(f" - {t['name']}({params}): {t['description']}") |
| 73 | descriptions[name] = "\n".join(lines) |
| 74 | except Exception as exc: # noqa: BLE001 |
| 75 | descriptions[name] = f" (unavailable: {exc})" |
| 76 | return descriptions |
| 77 | |
| 78 | async def execute_plan(self, plan: Plan, question: str) -> list[StepResult]: |
| 79 | """Execute all plan steps in dependency order.""" |