(args: dict[str, Any], ctx: Any)
| 124 | loop = self._loop |
| 125 | |
| 126 | def _call(args: dict[str, Any], ctx: Any) -> ToolResult: |
| 127 | try: |
| 128 | fut = asyncio.run_coroutine_threadsafe(client.call_tool(mcp_tool.name, args), loop) |
| 129 | res = fut.result(_CALL_TIMEOUT_S) |
| 130 | except Exception as exc: # noqa: BLE001 |
| 131 | return ToolResult(name=full, output=f"MCP call failed: {exc}", is_error=True) |
| 132 | return ToolResult( |
| 133 | name=full, |
| 134 | output=_render_content(getattr(res, "content", None)), |
| 135 | is_error=bool(getattr(res, "is_error", False)), |
| 136 | ) |
| 137 | |
| 138 | return build_tool( |
| 139 | name=full, |
nothing calls this directly
no test coverage detected