Python: Preserve provider error responses in hosting

python/packages/foundry_hosting/README.md

@@ -2,6 +2,14 @@
 
 This package provides the integration of Agent Framework agents and workflows with the Foundry Agent Server, which can be hosted on Foundry infrastructure.
 
+Chat-client exceptions retain their original SDK exception through the standard
+`inner_exception`/`__cause__` chain. Foundry Responses hosting copies the SDK's
+existing `code`, `message`, `status_code`, and body fields without interpreting
+the body or inferring an error title from HTTP status. Non-chat-client and
+malformed errors retain the generic sanitized fallback. Providers and
+application backends own the content and disclosure policy of the errors they
+return.
+
 ## Agent instances and factories
 
 `ResponsesHostServer` and `InvocationsHostServer` accept an agent instance or a zero-argument callable through the

python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py

@@ -52,7 +52,10 @@
 )
 from agent_framework._mcp import _MCP_TOOL_RESULT_HOST_PAYLOAD_KEY  # pyright: ignore[reportPrivateUsage]
 from agent_framework._telemetry import mark_feature_used
-from agent_framework.exceptions import AgentFrameworkException
+from agent_framework.exceptions import (
+    AgentFrameworkException,
+    ChatClientException,
+)
 from agent_framework.observability import OBSERVABILITY_SETTINGS
 from azure.ai.agentserver.core import get_request_context
 from azure.ai.agentserver.responses import (
@@ -149,6 +152,39 @@
 _AGENT_SOURCE_UNSET = object()
 
 
+def _provider_error_from_chat_client(exception: BaseException) -> dict[str, Any] | None:
+    """Copy the original SDK error already carried by a chat client exception."""
+    if not isinstance(exception, ChatClientException):
+        return None
+    inner: Exception | None = exception.__cause__ if isinstance(exception.__cause__, Exception) else None
+    if inner is None:
+        inner = next((arg for arg in exception.args[1:] if isinstance(arg, Exception)), None)
+    if inner is None:
+        return None
+    try:
+        code = getattr(inner, "code", None)
+        message = getattr(inner, "message", None)
+        status_code = getattr(inner, "status_code", None)
+        body = getattr(inner, "body", None)
+        if not isinstance(message, str) or not message:
+            return None
+        payload = (
+            {key: value for key, value in cast(Mapping[object, object], body).items() if isinstance(key, str)}
+            if isinstance(body, Mapping)
+            else {}
+        )
+        payload.setdefault("code", code if isinstance(code, str) and code else "server_error")
+        payload.setdefault("message", message)
+        if not isinstance(payload["code"], str) or not isinstance(payload["message"], str):
+            return None
+        if isinstance(status_code, int) and not isinstance(status_code, bool):
+            payload["status_code"] = status_code
+        json.dumps(payload, allow_nan=False)
+        return payload
+    except Exception:
+        return None
+
+
 class _PublicResponseFailure(Exception):
     """Mark a vetted failure message as safe for client-visible serialization."""
 
@@ -2349,11 +2385,12 @@ def _emit_failure(
         """Log ``ex`` and yield a terminal ``response.failed`` event.
 
         Drains any in-progress streaming output item first so the resulting
-        SSE stream stays well-formed, then emits ``response.failed`` carrying a
-        stable public message unless the caller explicitly supplies a safe
-        validation message. Any error raised while draining the tracker is
-        logged and otherwise ignored so that the original failure remains the
-        terminal failure.
+        SSE stream stays well-formed, then emits ``response.failed`` with a
+        supported HTTP classification and the originating provider message when
+        one is available. Vetted validation messages remain public, while
+        unexpected failures retain the stable fallback. Any tracker cleanup
+        error is logged and otherwise ignored so the original failure remains
+        terminal.
         """
         _log_response_failure(phase, response_id, ex)
         if tracker is not None:
@@ -2363,12 +2400,30 @@ def _emit_failure(
                 _log_response_failure("failed tracker cleanup", response_id, close_error)
             tracker.discard_pending_container_file_citations()
         if public_message is not None:
+            code = "server_error"
             message = public_message
         elif isinstance(ex, _PublicResponseFailure):
+            code = "server_error"
             message = ex.public_message
         else:
-            message = _PUBLIC_STREAM_FAILURE_MESSAGE
-        yield response_event_stream.emit_failed(message=message, usage=tracker.usage if tracker is not None else None)
+            provider_error = _provider_error_from_chat_client(ex)
+            code = cast(str, provider_error["code"]) if provider_error is not None else "server_error"
+            message = (
+                cast(str, provider_error["message"]) if provider_error is not None else _PUBLIC_STREAM_FAILURE_MESSAGE
+            )
+        failed_event = response_event_stream.emit_failed(
+            code=code, message=message, usage=tracker.usage if tracker is not None else None
+        )
+        if not isinstance(ex, _PublicResponseFailure) and isinstance(ex, ChatClientException):
+            provider_error = _provider_error_from_chat_client(ex)
+            if provider_error is not None:
+                failed_payload = cast(dict[str, Any], failed_event)
+                failed_response = cast(dict[str, Any], failed_payload["response"])
+                failed_error = cast(dict[str, Any], failed_response["error"])
+                failed_error.update(provider_error)
+                failed_error.setdefault("code", code)
+                failed_error.setdefault("message", message)
+        yield failed_event
 
 
 # endregion ResponsesHostServer

python/packages/foundry_hosting/tests/test_responses.py

@@ -60,6 +60,7 @@
 )
 from agent_framework._mcp import _MCP_TOOL_RESULT_HOST_PAYLOAD_KEY  # pyright: ignore[reportPrivateUsage]
 from agent_framework.ag_ui import AgentFrameworkAgent, InMemoryAGUIThreadSnapshotStore
+from agent_framework.exceptions import ChatClientException
 from agent_framework.observability import OBSERVABILITY_SETTINGS
 from agent_framework.openai import OpenAIChatClient, OpenAIChatOptions, OpenAIContinuationToken
 from azure.ai.agentserver.core import FoundryAgentRequestContext, get_request_context
@@ -119,6 +120,91 @@
 _SESSION_STORAGE_ERROR_DETAIL = "session-storage-secret at C:\\sensitive\\session_store.py"
 
 
+@pytest.mark.parametrize(
+    "status, code",
+    [
+        (400, "invalid_request_error"),
+        (403, "permission_denied"),
+        (429, "token_rate_limit_exceeded"),
+        (503, "backend_overloaded"),
+    ],
+)
+@pytest.mark.parametrize("wrapper", ["cause", "inner_exception"])
+def test_failure_event_classifies_provider_http_failures(status: int, code: str, wrapper: str) -> None:
+    class ProviderError(Exception):
+        status_code = status
+        message = "rendered SDK exception"
+
+        def __init__(self) -> None:
+            super().__init__(self.message)
+            self.code = code
+            self.body = {
+                "code": code,
+                "message": _PRIVATE_ERROR_DETAIL,
+                "type": "invalid_request_error",
+                "param": "input",
+                "request_id": "req_test",
+            }
+
+    provider_error = ProviderError()
+    if wrapper == "cause":
+        error = ChatClientException("provider wrapper")
+        error.__cause__ = provider_error
+    else:
+        error = ChatClientException("provider wrapper", inner_exception=provider_error)
+
+    stream = ResponseEventStream(response_id="resp_http_failure")
+    stream.emit_created()
+    stream.emit_in_progress()
+    events = list(
+        ResponsesHostServer._emit_failure(
+            "resp_http_failure",
+            "provider HTTP failure",
+            stream,
+            None,
+            error,
+        )
+    )
+    terminal = events[-1]
+    assert terminal["type"] == "response.failed"
+    failure = terminal["response"]["error"]
+    assert failure is not None
+    failure_mapping = cast(dict[str, Any], failure)
+    assert failure_mapping["code"] == code
+    assert failure_mapping["message"] == _PRIVATE_ERROR_DETAIL
+    assert failure_mapping["type"] == "invalid_request_error"
+    assert failure_mapping["param"] == "input"
+    assert failure_mapping["request_id"] == "req_test"
+    assert failure_mapping["status_code"] == status
+
+
+@pytest.mark.parametrize("unsafe_value", [b"not-json", object(), float("nan")])
+def test_failure_event_rejects_non_json_provider_body(unsafe_value: object) -> None:
+    class ProviderError(Exception):
+        code = "provider_error"
+        message = _PRIVATE_ERROR_DETAIL
+        status_code = 500
+        body = {"unsafe": unsafe_value}
+
+    error = ChatClientException("provider wrapper", inner_exception=ProviderError())
+    stream = ResponseEventStream(response_id="resp_unsafe_provider_error")
+    stream.emit_created()
+    stream.emit_in_progress()
+    events = list(
+        ResponsesHostServer._emit_failure(
+            "resp_unsafe_provider_error",
+            "unsafe provider failure",
+            stream,
+            None,
+            error,
+        )
+    )
+    terminal = cast(dict[str, Any], events[-1])
+    response = cast(dict[str, Any], terminal["response"])
+    failure = response["error"]
+    assert failure == {"code": "server_error", "message": "An internal server error occurred."}
+
+
 @pytest.fixture(scope="session", autouse=True)
 def _isolate_agentserver_state_root(tmp_path_factory: pytest.TempPathFactory) -> Generator[None]:
     """Keep the real local state store, but prevent xdist workers sharing its files."""

python/packages/hosting-responses/README.md

@@ -34,6 +34,13 @@ Missing counters never borrow from another field or become invented zeros; an
 absent total alone is derived from known input and output counts. Usage that
 cannot form a consistent Responses shape is omitted.
 
+Chat-client exceptions retain their original SDK exception through the standard
+`inner_exception`/`__cause__` chain. Responses hosting copies the SDK's existing
+`code`, `message`, `status_code`, and body fields without interpreting the body
+or inferring an error title from HTTP status. Non-chat-client and malformed
+errors retain the generic sanitized fallback. Providers and application
+backends own the content and disclosure policy of the errors they return.
+
 FastAPI/Starlette/Django/Azure Functions code owns route registration,
 authentication, status codes, response construction, and background work.
 

python/packages/hosting-responses/agent_framework_hosting_responses/_parsing.py

@@ -32,6 +32,7 @@
     get_namespaced_value,
 )
 from agent_framework._telemetry import mark_feature_used
+from agent_framework.exceptions import ChatClientException
 from agent_framework.observability import OBSERVABILITY_SETTINGS
 from agent_framework_hosting import AgentRunArgs
 from openai.types.responses import (
@@ -81,6 +82,39 @@
 _STREAMING_TERMINAL_STATUSES = frozenset({"completed", "failed", "incomplete"})
 
 
+def _provider_error_from_chat_client(exception: Exception) -> dict[str, Any] | None:
+    """Copy the original SDK error already carried by a chat client exception."""
+    if not isinstance(exception, ChatClientException):
+        return None
+    inner: Exception | None = exception.__cause__ if isinstance(exception.__cause__, Exception) else None
+    if inner is None:
+        inner = next((arg for arg in exception.args[1:] if isinstance(arg, Exception)), None)
+    if inner is None:
+        return None
+    try:
+        code = getattr(inner, "code", None)
+        message = getattr(inner, "message", None)
+        status_code = getattr(inner, "status_code", None)
+        body = getattr(inner, "body", None)
+        if not isinstance(message, str) or not message:
+            return None
+        payload = (
+            {key: value for key, value in cast(Mapping[object, object], body).items() if isinstance(key, str)}
+            if isinstance(body, Mapping)
+            else {}
+        )
+        payload.setdefault("code", code if isinstance(code, str) and code else "server_error")
+        payload.setdefault("message", message)
+        if not isinstance(payload["code"], str) or not isinstance(payload["message"], str):
+            return None
+        if isinstance(status_code, int) and not isinstance(status_code, bool):
+            payload["status_code"] = status_code
+        json.dumps(payload, allow_nan=False)
+        return payload
+    except Exception:
+        return None
+
+
 def _log_stream_failure(response_id: str, exc: Exception) -> None:
     """Log a correlated stream failure under the framework's sensitive-data policy."""
     if OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED:
@@ -1523,6 +1557,11 @@ def apply_streamed_message_ids(response: dict[str, Any]) -> None:
         partial_contents = [
             content for update in updates for content in update.contents if content.type == "text" and content.text
         ]
+        provider_error = _provider_error_from_chat_client(exc)
+        error_code = cast(str, provider_error["code"]) if provider_error is not None else "server_error"
+        error_message = (
+            cast(str, provider_error["message"]) if provider_error is not None else _PUBLIC_STREAM_FAILURE_MESSAGE
+        )
         response_kwargs: dict[str, Any] = {
             "id": response_id,
             "object": "response",
@@ -1535,13 +1574,20 @@ def apply_streamed_message_ids(response: dict[str, Any]) -> None:
             "tools": [],
             "metadata": {},
             "error": {
+                # The OpenAI SDK model currently restricts error codes to its
+                # own known literals; validate the response shape first, then
+                # restore the provider-defined structured code below.
                 "code": "server_error",
-                "message": _PUBLIC_STREAM_FAILURE_MESSAGE,
+                "message": error_message,
             },
         }
         if conversation_id is not None:
             response_kwargs["conversation"] = {"id": conversation_id}
         failed_response = _response_payload(OpenAIResponse(**response_kwargs))
+        if provider_error is not None:
+            failed_response["error"].update(provider_error)
+        failed_response["error"].setdefault("code", error_code)
+        failed_response["error"].setdefault("message", error_message)
         apply_streamed_message_ids(failed_response)
         yield event(
             "response.failed",