Python: Convert nested SDK models in hosted tool sanitization

python/packages/foundry/agent_framework_foundry/_tools.py

@@ -33,6 +33,19 @@ def _validate_hosted_tool_payload(sanitized: Mapping[str, Any]) -> None:
         )
 
 
+def _to_plain_dict(value: Any) -> Any:
+    """Recursively convert Azure SDK models and mappings into plain dictionaries."""
+    if isinstance(value, Mapping):
+        mapping = cast("Mapping[str, Any]", value)
+        return {k: _to_plain_dict(v) for k, v in mapping.items()}
+    if hasattr(value, "as_dict") and callable(value.as_dict):
+        return _to_plain_dict(value.as_dict())
+    if isinstance(value, (list, tuple)):
+        items = cast("list[Any] | tuple[Any, ...]", value)
+        return [_to_plain_dict(v) for v in items]
+    return value
+
+
 def _sanitize_foundry_response_tool(tool_item: Any) -> Any:  # pyright: ignore[reportUnusedFunction]
     """Return a Responses-API-safe tool payload for Foundry hosted tools.
 
@@ -50,9 +63,12 @@ def _sanitize_foundry_response_tool(tool_item: Any) -> Any:  # pyright: ignore[r
        defaulted (``file_search`` without ``vector_store_ids``, ``mcp`` without
        either ``server_url`` or ``project_connection_id``) raise ``ValueError``
        with a message that points at the tool definition.
+    4. Nested Azure SDK models (such as ``BingCustomSearchToolParameters`` or
+       ``AutoCodeInterpreterToolParam``) are recursively converted into plain
+       dictionaries so the resulting payload is JSON serializable.
     """
     if isinstance(tool_item, FoundryMCPTool):
-        sanitized: dict[str, Any] = dict(cast("Mapping[str, Any]", tool_item))
+        sanitized: dict[str, Any] = _to_plain_dict(tool_item)
         sanitized.pop("name", None)
         sanitized.pop("description", None)
         _validate_hosted_tool_payload(sanitized)
@@ -61,7 +77,7 @@ def _sanitize_foundry_response_tool(tool_item: Any) -> Any:  # pyright: ignore[r
     if isinstance(tool_item, Mapping):
         mapping = cast("Mapping[str, Any]", tool_item)
         if "type" in mapping and mapping.get("type") not in {"function", "custom"}:
-            sanitized = dict(mapping)
+            sanitized = _to_plain_dict(mapping)
             sanitized.pop("name", None)
             sanitized.pop("description", None)
             if sanitized.get("type") == "code_interpreter" and "container" not in sanitized:

python/packages/foundry/tests/foundry/test_foundry_chat_client.py

@@ -8,6 +8,7 @@
 import os
 import sys
 import warnings
+from collections.abc import Iterator, Mapping
 from functools import wraps
 from importlib import import_module
 from pathlib import Path
@@ -47,6 +48,7 @@
 
 from agent_framework_foundry import FoundryChatClient, RawFoundryChatClient
 from agent_framework_foundry._feature_usage import FeatureIndex, FeatureUsagePolicy
+from agent_framework_foundry._tools import _to_plain_dict
 
 _OPENAI_HTTPX = cast(Any, import_module(DefaultAsyncHttpxClient.__mro__[1].__module__.partition(".")[0]))
 
@@ -1336,6 +1338,157 @@ def test_prepare_tools_for_openai_strips_name_from_non_function_hosted_tool_dict
     assert "description" not in response_tools[1]
 
 
+def test_prepare_tools_for_openai_serializes_bing_custom_search_tool() -> None:
+    """Hosted Bing custom search tool contains nested models that must be JSON serializable."""
+    project_client = MagicMock()
+    project_client.get_openai_client.return_value = _make_mock_openai_client()
+    client = FoundryChatClient(project_client=project_client, model="test-model")
+
+    tool = FoundryChatClient.get_bing_custom_search_tool(
+        connection_id="/subscriptions/x/connections/bing",
+        instance_name="inst",
+        market="zh-TW",
+    )
+
+    response_tools = client._prepare_tools_for_openai([tool])
+
+    assert len(response_tools) == 1
+    # Ensure nested Azure SDK models were converted and are JSON serializable
+    dumped = json.dumps(response_tools)
+    assert "bing_custom_search_preview" in dumped
+    assert "project_connection_id" in dumped
+    assert "inst" in dumped
+
+
+def test_prepare_tools_for_openai_serializes_code_interpreter_with_auto_param() -> None:
+    """Hosted code interpreter tool from factory must be JSON serializable."""
+    project_client = MagicMock()
+    project_client.get_openai_client.return_value = _make_mock_openai_client()
+    client = FoundryChatClient(project_client=project_client, model="test-model")
+
+    tool = FoundryChatClient.get_code_interpreter_tool()
+    response_tools = client._prepare_tools_for_openai([tool])
+
+    assert len(response_tools) == 1
+    dumped = json.dumps(response_tools)
+    assert "code_interpreter" in dumped
+    assert "auto" in dumped
+
+
+def test_to_plain_dict_prefers_mapping_over_incompatible_as_dict() -> None:
+    """Mapping objects with an incompatible as_dict method are converted without calling as_dict."""
+
+    class IncompatibleMapping(Mapping[str, Any]):
+        def __init__(self, data: dict[str, Any]) -> None:
+            self._data = data
+            self.as_dict_called = False
+
+        def __getitem__(self, key: str) -> Any:
+            return self._data[key]
+
+        def __iter__(self) -> Iterator[str]:
+            return iter(self._data)
+
+        def __len__(self) -> int:
+            return len(self._data)
+
+        def as_dict(self, required_arg: Any) -> dict[str, Any]:
+            self.as_dict_called = True
+            raise TypeError("as_dict() missing 1 required positional argument: 'required_arg'")
+
+    class MockAzureSdkModel:
+        def __init__(self, payload: dict[str, Any]) -> None:
+            self._payload = payload
+            self.as_dict_called = False
+
+        def as_dict(self) -> dict[str, Any]:
+            self.as_dict_called = True
+            return self._payload
+
+    nested_mapping = IncompatibleMapping({"inner_key": "inner_value"})
+    sdk_model = MockAzureSdkModel({"sdk_key": "sdk_value"})
+    nested_list_item = IncompatibleMapping({"list_key": 42})
+
+    top_level = IncompatibleMapping({
+        "type": "custom_hosted_tool",
+        "nested": nested_mapping,
+        "sdk": sdk_model,
+        "items": [nested_list_item],
+    })
+
+    result = _to_plain_dict(top_level)
+
+    # Verify that the result and nested containers are regular plain dictionaries
+    assert type(result) is dict
+    assert type(result["nested"]) is dict
+    assert type(result["sdk"]) is dict
+    assert type(result["items"][0]) is dict
+
+    # Verify values are recursively converted
+    assert result == {
+        "type": "custom_hosted_tool",
+        "nested": {"inner_key": "inner_value"},
+        "sdk": {"sdk_key": "sdk_value"},
+        "items": [{"list_key": 42}],
+    }
+
+    # Verify that as_dict was never invoked on mapping instances
+    assert not top_level.as_dict_called
+    assert not nested_mapping.as_dict_called
+    assert not nested_list_item.as_dict_called
+
+    # Verify that non-mapping Azure SDK model as_dict was invoked
+    assert sdk_model.as_dict_called
+
+
+def test_prepare_tools_for_openai_handles_mapping_with_incompatible_as_dict() -> None:
+    """Hosted tool mapping with an incompatible as_dict method is sanitized and JSON serializable."""
+
+    class IncompatibleToolMapping(Mapping[str, Any]):
+        def __init__(self, data: dict[str, Any]) -> None:
+            self._data = data
+            self.as_dict_called = False
+
+        def __getitem__(self, key: str) -> Any:
+            return self._data[key]
+
+        def __iter__(self) -> Iterator[str]:
+            return iter(self._data)
+
+        def __len__(self) -> int:
+            return len(self._data)
+
+        def as_dict(self, required_arg: Any) -> dict[str, Any]:
+            self.as_dict_called = True
+            raise TypeError("as_dict() missing 1 required positional argument: 'required_arg'")
+
+    project_client = MagicMock()
+    project_client.get_openai_client.return_value = _make_mock_openai_client()
+    client = FoundryChatClient(project_client=project_client, model="test-model")
+
+    tool = IncompatibleToolMapping({
+        "type": "code_interpreter",
+        "container": {"type": "auto"},
+        "name": "read_model_name",
+        "description": "read_model_desc",
+    })
+
+    response_tools = client._prepare_tools_for_openai([tool])
+
+    assert len(response_tools) == 1
+    sanitized = response_tools[0]
+    assert type(sanitized) is dict
+    assert "name" not in sanitized
+    assert "description" not in sanitized
+    assert sanitized["type"] == "code_interpreter"
+    assert sanitized["container"] == {"type": "auto"}
+    assert not tool.as_dict_called
+
+    dumped = json.dumps(response_tools)
+    assert "code_interpreter" in dumped
+    assert "auto" in dumped
+
+
 @pytest.mark.flaky
 @pytest.mark.integration
 @skip_if_foundry_integration_tests_disabled