diff --git a/spark_executor/server.py b/spark_executor/server.py index 377e66f..a8ae214 100644 --- a/spark_executor/server.py +++ b/spark_executor/server.py @@ -557,7 +557,15 @@ def _update_job_file(req: UpdateJobFileRequest): "Connection's saved auth is reused, so the agent does not need cluster " "credentials.\n\n" "**Limits:** 30s timeout, redirects followed, response body capped at " - "1 MB (the response includes a `truncated` boolean when this kicks in)." + "1 MB (the response includes a `truncated` boolean when this kicks in).\n\n" + "**Errors:** when the host is unreachable (connect refused, DNS " + "failure, TLS handshake error, timeout, etc.) this tool returns " + "HTTP 400 with the exception class and message in the response " + "detail — e.g. `fetch_url could not reach ...: ConnectError: " + "Connection refused`. Upstream HTTP 4xx/5xx responses that DID " + "come back are returned in the result body with their status code " + "preserved (not translated to an error) so you can see what the " + "server actually said." ), ) def _fetch_url(req: FetchUrlRequest): diff --git a/spark_executor/tools/fetch_url.py b/spark_executor/tools/fetch_url.py index 9615372..d80243a 100644 --- a/spark_executor/tools/fetch_url.py +++ b/spark_executor/tools/fetch_url.py @@ -83,13 +83,27 @@ def fetch_url(url: str, connection_name: str) -> FetchUrlResult: verify = config.verify_for_httpx() for hop in range(_MAX_REDIRECTS + 1): - resp = httpx.get( - url, - auth=auth, - verify=verify, - timeout=_REQUEST_TIMEOUT_SECONDS, - follow_redirects=False, - ) + try: + resp = httpx.get( + url, + auth=auth, + verify=verify, + timeout=_REQUEST_TIMEOUT_SECONDS, + follow_redirects=False, + ) + except httpx.HTTPError as exc: + # Network-level failure (no response received). Surface the + # exception class + message so the LLM can act on it. + # ValueError -> 400 via the existing handler in server.py. + # The cause chain (`from exc`) preserves the original + # exception for loguru. + raise ValueError( + f"fetch_url could not reach {url!r}: " + f"{type(exc).__name__}: {exc}. " + f"Check that the URL is reachable from the MCP service, " + f"the host is in Connection.url_allowlist, and the " + f"connection's auth/SSL settings are correct." + ) from exc if resp.status_code not in _REDIRECT_STATUSES: logger.info( f"fetch_url ok url={url} connection_name={connection_name} " diff --git a/tests/unit/test_fetch_url.py b/tests/unit/test_fetch_url.py index affe030..c26cfe0 100644 --- a/tests/unit/test_fetch_url.py +++ b/tests/unit/test_fetch_url.py @@ -327,3 +327,76 @@ def test_list_applications_request_limit_bounds(): ListApplicationsRequest(connection_name="prod", limit=0) with pytest.raises(ValidationError): ListApplicationsRequest(connection_name="prod", limit=10001) + + +# --- Network error handling (httpx.HTTPError -> 400 with detail) --- + + +def test_fetch_url_raises_400_with_detail_on_connect_error(fresh_stores): + """When httpx.get raises ConnectError (host unreachable / port closed), + fetch_url must raise ValueError (-> 400) with the exception class + and message in the detail. Previously this bubbled up as a bare + 500 Internal Server Error with no info.""" + fetch_url.conn_store.save( + Connection( + name="prod", + master="yarn", + yarn_rm_url="http://rm.prod.internal:8088", + url_allowlist=["*.prod.internal"], + ) + ) + with patch( + "spark_executor.tools.fetch_url.httpx.get", + side_effect=httpx.ConnectError("Connection refused"), + ): + with pytest.raises(ValueError) as ei: + fetch_url.fetch_url("http://nm01.prod.internal:8042/", "prod") + msg = str(ei.value) + assert "ConnectError" in msg + assert "Connection refused" in msg + assert "nm01.prod.internal" in msg + # Cause chain preserved for loguru + assert isinstance(ei.value.__cause__, httpx.ConnectError) + + +def test_fetch_url_raises_400_with_detail_on_timeout(fresh_stores): + """When httpx.get raises TimeoutException (request exceeded 30s), + fetch_url must raise ValueError with the exception details surfaced.""" + fetch_url.conn_store.save( + Connection( + name="prod", + master="yarn", + yarn_rm_url="http://rm.prod.internal:8088", + url_allowlist=["*.prod.internal"], + ) + ) + with patch( + "spark_executor.tools.fetch_url.httpx.get", + side_effect=httpx.ReadTimeout("Timed out reading"), + ): + with pytest.raises(ValueError) as ei: + fetch_url.fetch_url("http://nm01.prod.internal:8042/", "prod") + msg = str(ei.value) + assert "ReadTimeout" in msg + assert "Timed out reading" in msg + assert isinstance(ei.value.__cause__, httpx.ReadTimeout) + + +def test_fetch_url_returns_body_for_4xx_5xx_upstream(fresh_stores): + """Upstream HTTP errors (4xx/5xx responses that DID come back) are + NOT translated to ValueError — the FetchUrlResult carries the status + code and body so the LLM can see what the server actually said. This + is the intentional contrast with the no-response-at-all case.""" + fetch_url.conn_store.save( + Connection( + name="prod", + master="yarn", + yarn_rm_url="http://rm.prod.internal:8088", + url_allowlist=["*.prod.internal"], + ) + ) + resp = httpx.Response(503, text="Service Unavailable - try again later") + with patch("spark_executor.tools.fetch_url.httpx.get", return_value=resp): + out = fetch_url.fetch_url("http://nm01.prod.internal:8042/", "prod") + assert out.status_code == 503 + assert "Service Unavailable" in out.body