fix(tools): send OpenAPI operations to their own declared servers

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(

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."