Fetch available tools from the MCP server.
(self)
| 308 | # ── Public API ─────────────────────────────────────────────── |
| 309 | |
| 310 | async def list_tools(self) -> list[dict]: |
| 311 | """Fetch available tools from the MCP server.""" |
| 312 | try: |
| 313 | data = await self._detect_and_request("tools/list") |
| 314 | |
| 315 | if "error" in data: |
| 316 | err = data["error"] |
| 317 | msg = err.get("message", str(err)) if isinstance(err, dict) else str(err) |
| 318 | raise Exception(f"MCP error: {msg}") |
| 319 | |
| 320 | result = data.get("result", {}) |
| 321 | tools = result.get("tools", []) if isinstance(result, dict) else [] |
| 322 | return [ |
| 323 | { |
| 324 | "name": t.get("name", ""), |
| 325 | "description": t.get("description", ""), |
| 326 | "inputSchema": t.get("inputSchema", {}), |
| 327 | } |
| 328 | for t in tools |
| 329 | ] |
| 330 | except httpx.HTTPError as e: |
| 331 | raise Exception(f"Connection failed: {str(e)[:200]}") |
| 332 | |
| 333 | async def call_tool(self, tool_name: str, arguments: dict) -> str: |
| 334 | """Execute a tool on the MCP server.""" |