From 518500cb7add61a76000310b7063146062cbcfa3 Mon Sep 17 00:00:00 2001 From: Julien Parenti Date: Tue, 10 Nov 2026 22:54:11 +0100 Subject: [PATCH] fix(tools): send OpenAPI operations to their own declared servers OpenApiSpecParser picked the root servers[0].url once and reused it for every operation, so a path item or operation that declared its own servers was still called on the root server. Per the OpenAPI spec, operation-level servers override path-level servers, which override the root servers. Resolve the base URL per operation in that order; an empty servers list does not override the enclosing level. Fixes #8064 --- .../openapi_spec_parser.py | 22 +++++- .../test_openapi_spec_parser.py | 73 +++++++++++++++++++ 2 files changed, 92 insertions(+), 3 deletions(-) diff --git a/src/google/adk/tools/openapi_tool/openapi_spec_parser/openapi_spec_parser.py b/src/google/adk/tools/openapi_tool/openapi_spec_parser/openapi_spec_parser.py index 1745698e410..8b91ec51c87 100644 --- a/src/google/adk/tools/openapi_tool/openapi_spec_parser/openapi_spec_parser.py +++ b/src/google/adk/tools/openapi_tool/openapi_spec_parser/openapi_spec_parser.py @@ -47,6 +47,14 @@ _SCHEMA_CONTAINER_KEYS: Set[str] = frozenset({"schema", "schemas"}) +def _first_server_url(servers: list[dict[str, Any]] | None) -> str | None: + """Returns the URL of the first server, or None if no server is declared.""" + if not servers: + return None + url: str = servers[0].get("url", "") + return url + + class OperationEndpoint(BaseModel): base_url: str path: str @@ -178,9 +186,7 @@ def _collect_operations( operations = [] # Taking first server url, or default to empty string if not present - base_url = "" - if openapi_spec.get("servers"): - base_url = openapi_spec["servers"][0].get("url", "") + root_base_url = _first_server_url(openapi_spec.get("servers")) or "" # Get global security scheme (if any). global_scheme_name = _required_scheme_name(openapi_spec.get("security")) @@ -191,6 +197,11 @@ def _collect_operations( if path_item is None: continue + # Path-level servers override the root servers for this path. + path_base_url = _first_server_url(path_item.get("servers")) + if path_base_url is None: + path_base_url = root_base_url + for method in ( "get", "post", @@ -227,6 +238,11 @@ def _collect_operations( temp_id = _to_snake_case(f"{path}_{method}") operation_dict["operationId"] = temp_id + # Operation-level servers override the path and root servers. + base_url = _first_server_url(operation_dict.get("servers")) + if base_url is None: + base_url = path_base_url + url = OperationEndpoint(base_url=base_url, path=path, method=method) operation = Operation.model_validate(operation_dict) operation_parser = OperationParser( diff --git a/tests/unittests/tools/openapi_tool/openapi_spec_parser/test_openapi_spec_parser.py b/tests/unittests/tools/openapi_tool/openapi_spec_parser/test_openapi_spec_parser.py index f8f29d1a968..bd6f6dc35a2 100644 --- a/tests/unittests/tools/openapi_tool/openapi_spec_parser/test_openapi_spec_parser.py +++ b/tests/unittests/tools/openapi_tool/openapi_spec_parser/test_openapi_spec_parser.py @@ -12,6 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. +import copy from typing import Any from typing import Dict @@ -306,6 +307,78 @@ def test_parse_spec_with_no_servers(openapi_spec_generator): assert parsed_operations[0].endpoint.base_url == "" +def test_operation_servers_override_path_and_root_servers( + openapi_spec_generator, +): + """An operation that declares servers is sent to its own server.""" + openapi_spec = create_minimal_openapi_spec() + openapi_spec["servers"] = [{"url": "https://root.example.com"}] + path_item = openapi_spec["paths"]["/test"] + path_item["servers"] = [{"url": "https://path.example.com"}] + path_item["get"]["servers"] = [ + {"url": "https://operation.example.com"}, + {"url": "https://operation-fallback.example.com"}, + ] + + parsed_operations = openapi_spec_generator.parse(openapi_spec) + + assert len(parsed_operations) == 1 + assert ( + parsed_operations[0].endpoint.base_url == "https://operation.example.com" + ) + + +def test_path_servers_override_root_servers(openapi_spec_generator): + """Operations without servers of their own use their path's servers.""" + openapi_spec = create_minimal_openapi_spec() + openapi_spec["servers"] = [{"url": "https://root.example.com"}] + openapi_spec["paths"]["/test"]["servers"] = [ + {"url": "https://path.example.com"} + ] + openapi_spec["paths"]["/other"] = copy.deepcopy( + openapi_spec["paths"]["/test"] + ) + del openapi_spec["paths"]["/other"]["servers"] + openapi_spec["paths"]["/other"]["get"]["operationId"] = "otherGet" + + parsed_operations = openapi_spec_generator.parse(openapi_spec) + + base_urls = {op.name: op.endpoint.base_url for op in parsed_operations} + assert base_urls == { + "test_get": "https://path.example.com", + "other_get": "https://root.example.com", + } + + +def test_path_servers_apply_when_no_root_servers(openapi_spec_generator): + """Path-level servers are used even when the spec has no root servers.""" + openapi_spec = create_minimal_openapi_spec() + openapi_spec["paths"]["/test"]["servers"] = [ + {"url": "https://path.example.com"} + ] + + parsed_operations = openapi_spec_generator.parse(openapi_spec) + + assert len(parsed_operations) == 1 + assert parsed_operations[0].endpoint.base_url == "https://path.example.com" + + +def test_empty_operation_servers_fall_back_to_path_servers( + openapi_spec_generator, +): + """An empty servers list on an operation does not override its path.""" + openapi_spec = create_minimal_openapi_spec() + openapi_spec["servers"] = [{"url": "https://root.example.com"}] + path_item = openapi_spec["paths"]["/test"] + path_item["servers"] = [{"url": "https://path.example.com"}] + path_item["get"]["servers"] = [] + + parsed_operations = openapi_spec_generator.parse(openapi_spec) + + assert len(parsed_operations) == 1 + assert parsed_operations[0].endpoint.base_url == "https://path.example.com" + + def test_parse_spec_with_description(openapi_spec_generator): openapi_spec = create_minimal_openapi_spec() expected_description = "This is a test description."