Python: Keep tool diagnostics internal by default

docs/specs/004-python-function-calling-loop.md

@@ -369,6 +369,10 @@ that manually replay messages own the equivalent rule: do not resend an approval
   already executing in a worker thread cannot be interrupted and may complete its side effects — its result is
   discarded either way and never reaches the transcript, the model, or history. Middleware must not catch
   `MiddlewareFailure` — swallowing it converts a fail-closed abort back into a running, possibly unguarded loop.
+- `Content.exception` is host-internal diagnostic state. Default `Content.to_dict()` and nested response serialization replace it with a fixed non-sensitive failure marker,
+  while the original field remains directly available to trusted local code. Remote protocol serializers use the
+  marker only for status and use the channel-visible `result` or `items` for output text. `include_detailed_errors=False` keeps the channel-visible
+  result generic; enabling it explicitly may place diagnostic text in the result for that configured channel.
 - Parallel calls retain model order in the returned transcript.
 - `call_id` remains the provider/service correlation id; a locally actionable `function_call` also carries a stable
   Agent Framework occurrence identity in `Content.id`.
@@ -582,6 +586,7 @@ that manually replay messages own the equivalent rule: do not resend an approval
 |---|---|---|
 | Rejected execution | Rejection is a normal terminal result, not an exception to the caller. | `test_unapproved_tool_execution_raises_exception` |
 | Approved tool exception | Generic and detailed error modes preserve one result and one execution. | `test_approved_function_call_with_error_without_detailed_errors`, `test_approved_function_call_with_error_with_detailed_errors` |
+| Tool exception diagnostics | Internal diagnostics remain available to trusted local code, serialization preserves only a fixed failure marker, and explicit detailed-error configuration affects only the channel-visible result. | `packages/core/tests/core/test_types.py::test_function_result_exception_is_internal_by_default`, `packages/core/tests/core/test_function_invocation_logic.py::test_function_invocation_config_include_detailed_errors_false`, `test_function_invocation_config_include_detailed_errors_true`, `test_streaming_function_invocation_config_include_detailed_errors_false`, `test_streaming_function_invocation_config_include_detailed_errors_true` |
 | Approved validation error | Validation failure returns one result without invoking the function body. | `test_approved_function_call_with_validation_error` |
 | Approved success | Successful approved execution returns one result. | `test_approved_function_call_successful_execution` |
 | Consecutive error cap | Error threshold stops repeated failures, submits collected results, and makes only the required final no-tool model call. | `test_function_invocation_config_max_consecutive_errors`, `test_streaming_function_invocation_config_max_consecutive_errors`, `test_approval_resume_error_limit_forces_final_no_tool_response` |
@@ -611,6 +616,7 @@ that manually replay messages own the equivalent rule: do not resend an approval
 | OpenAI end-to-end hosted approval | Hosted request parses, response sends, and continuation completes. | `test_end_to_end_mcp_approval_flow` |
 | Stored function call/result | Service-side storage drops server-issued calls but keeps new outputs. | `test_prepare_options_with_conversation_id_strips_server_issued_items`, `test_prepare_messages_for_openai_full_conversation_with_reasoning` |
 | Stateless reasoning replay | Replay reconstructs reasoning, call, and result together; missing required reasoning fails before the request. | `test_tool_loop_store_false_replays_encrypted_reasoning_group`, `test_stateless_request_rejects_non_replayable_reasoning_bound_mcp_output`, `test_prepare_messages_for_openai_full_conversation_with_reasoning` |
+| Remote tool error serialization | Foundry Responses, Responses hosting, and AG-UI preserve channel-visible results without host-internal diagnostics. A2A and MCP hosting omit unsupported intermediate function results entirely, including their diagnostics. | `packages/foundry_hosting/tests/test_responses.py::TestNonStreaming::test_function_result_omits_internal_exception`, `TestStreaming::test_function_result_omits_internal_exception`, `packages/hosting-responses/tests/hosting_responses/test_parsing.py::TestResponsesRunHelpers::test_responses_from_run_omits_internal_function_exception`, `test_responses_from_streaming_run_omits_internal_function_exception`, `packages/ag-ui/tests/ag_ui/test_run_common.py::TestEmitToolResult::test_tool_result_does_not_emit_internal_exception`, `packages/hosting-a2a/tests/hosting_a2a/test_conversion.py::test_a2a_from_run_omits_unsupported_content`, `packages/hosting-mcp/tests/hosting_mcp/test_conversion.py::test_mcp_from_run_omits_content_not_supported_in_tool_results` |
 | Foundry encrypted reasoning opt-in | Foundry clients omit `reasoning.encrypted_content` by default and preserve an explicit caller opt-in. | `packages/foundry/tests/foundry/test_foundry_chat_client.py::test_get_response_does_not_request_encrypted_reasoning_by_default`, `test_get_response_preserves_explicit_encrypted_reasoning_opt_in`, `packages/foundry/tests/foundry/test_foundry_agent.py::test_foundry_agent_basic_call_does_not_request_unsupported_encrypted_reasoning`, `test_foundry_agent_preserves_caller_requested_encrypted_reasoning`, `packages/foundry_hosting/tests/test_responses_int.py::TestReasoningHostedMcpReplay::test_second_turn_replays_mcp_call_with_encrypted_reasoning` |
 | Opaque reasoning signature replay | Provider-specific opaque reasoning metadata is captured and restored on reconstructed calls. | `packages/gemini/tests/test_gemini_client.py::test_function_call_part_captures_thought_signature_as_reasoning_content`, `test_reconstructed_function_call_replays_thought_signature_from_reasoning_content` |
 | Chat Completions approval wrappers | Framework approval wrappers are not sent as chat messages. | `packages/openai/tests/openai/test_openai_chat_completion_client.py` approval serialization tests |

python/packages/ag-ui/tests/ag_ui/test_run_common.py

@@ -253,6 +253,27 @@ def test_tool_result_closes_open_text_message(self):
         assert flow.message_id is None
         assert flow.accumulated_text == ""
 
+    def test_tool_result_does_not_emit_internal_exception(self):
+        """AG-UI events and snapshots contain only the channel-visible result."""
+        diagnostic = "test-token-value at /srv/private/tool.py"
+        content = Content.from_function_result(
+            call_id="call_1",
+            result="Error: Function failed.",
+            exception=diagnostic,
+        )
+        flow = FlowState()
+
+        events = _emit_tool_result(content, flow)
+        payload = json.dumps(
+            {
+                "events": [event.model_dump(mode="json", by_alias=True) for event in events],
+                "snapshot": flow.tool_results,
+            }
+        )
+
+        assert "Error: Function failed." in payload
+        assert diagnostic not in payload
+
 
 class TestStateUpdateHelper:
     """Tests for the public ``state_update`` helper."""

python/packages/anthropic/tests/test_anthropic_client.py

@@ -374,6 +374,31 @@ def test_prepare_message_for_anthropic_function_result(
     assert result["content"][0]["is_error"] is False
 
 
+def test_prepare_message_for_anthropic_preserves_redacted_error_status(
+    mock_anthropic_client: MagicMock,
+) -> None:
+    """Persisted tool failures retain error status without exposing diagnostics."""
+    client = create_test_anthropic_client(mock_anthropic_client)
+    diagnostic = "test-token-value at /srv/private/tool.py"
+    failed_result = Content.from_function_result(
+        call_id="call_123",
+        result="Error: Function failed.",
+        exception=diagnostic,
+    )
+    restored_result = Content.from_dict(failed_result.to_dict())
+    message = Message(role="tool", contents=[restored_result])
+
+    result = client._prepare_message_for_anthropic(message)
+
+    tool_result = result["content"][0]
+    assert tool_result["type"] == "tool_result"
+    assert tool_result["is_error"] is True
+    assert diagnostic not in str(tool_result)
+    tool_content = tool_result["content"]
+    assert isinstance(tool_content, list)
+    assert tool_content[0]["text"] == "Error: Function failed."
+
+
 def test_prepare_message_for_anthropic_function_result_with_data_image(
     mock_anthropic_client: MagicMock,
 ) -> None:

python/packages/bedrock/agent_framework_bedrock/_chat_client.py

@@ -572,24 +572,13 @@ def _convert_content_to_bedrock_block(self, content: Content) -> dict[str, Any]
                     tool_result_blocks = self._convert_tool_result_to_blocks(tool_result_text)
                 else:
                     tool_result_blocks = self._convert_tool_result_to_blocks(content.result)
-                tool_result_block = {
+                return {
                     "toolResult": {
                         "toolUseId": content.call_id,
                         "content": tool_result_blocks,
-                        "status": "error" if content.exception else "success",
+                        "status": "error" if content.exception is not None else "success",
                     }
                 }
-                if content.exception:
-                    tool_result = tool_result_block["toolResult"]
-                    existing_content = tool_result.get("content")
-                    content_list: list[dict[str, Any]]
-                    if isinstance(existing_content, list):
-                        content_list = existing_content
-                    else:
-                        content_list = []
-                        tool_result["content"] = content_list
-                    content_list.append({"text": str(content.exception)})
-                return tool_result_block
             case _:
                 # Bedrock does not support other content types at this time
                 pass

python/packages/bedrock/tests/test_bedrock_client.py

@@ -448,24 +448,34 @@ def test_align_tool_results_handles_pending_edge_cases() -> None:
 def test_convert_content_to_bedrock_block_handles_errors_and_missing_items() -> None:
     """Function result conversion should serialize items, rich content warnings, and fallback results."""
     client = _make_client()
+    diagnostic = "test-token-value at /srv/private/tool.py"
     rich_result = Content.from_function_result(
         call_id="call-1",
         result=[Content.from_text(text="summary"), Content.from_data(data=b"x", media_type="image/png")],
-        exception="tool failed",
+        exception=diagnostic,
     )
+    restored_rich_result = Content.from_dict(rich_result.to_dict())
     fallback_result = Content.from_function_result(call_id="call-2", result={"answer": 42})
     fallback_result.items = None
 
-    rich_block = client._convert_content_to_bedrock_block(rich_result)
+    rich_block = client._convert_content_to_bedrock_block(restored_rich_result)
     fallback_block = client._convert_content_to_bedrock_block(fallback_result)
 
     assert rich_block == {
         "toolResult": {
             "toolUseId": "call-1",
-            "content": [{"text": "summary"}, {"text": "tool failed"}],
+            "content": [{"text": "summary"}],
             "status": "error",
         }
     }
+    assert diagnostic not in str(rich_block)
+
+    empty_diagnostic_block = client._convert_content_to_bedrock_block(
+        Content.from_function_result(call_id="call-empty", result="failed", exception="")
+    )
+    assert empty_diagnostic_block is not None
+    assert empty_diagnostic_block["toolResult"]["status"] == "error"
+
     assert fallback_block == {
         "toolResult": {
             "toolUseId": "call-2",

python/packages/core/agent_framework/_compaction.py

@@ -1273,8 +1273,8 @@ def _format_summary_content(content: Content) -> str:
         result_text = _format_summary_result_items(content.items) if content.items else ""
         if not result_text:
             result_text = _tool_result_text(content.result) if content.result is not None else "no result"
-        if content.exception:
-            result_text = f"error({content.exception}): {result_text}"
+        if content.exception is not None:
+            result_text = f"error: {result_text}"
         call_id_suffix = f" [call_id={content.call_id}]" if content.call_id else ""
         return f"function_result: {result_text}{call_id_suffix}"
     if content.type == "mcp_server_tool_call":
@@ -1285,8 +1285,8 @@ def _format_summary_content(content: Content) -> str:
         return call
     if content.type == "mcp_server_tool_result":
         result_text = _tool_result_text(content.output)
-        if content.exception:
-            result_text = f"error({content.exception}): {result_text}"
+        if content.exception is not None:
+            result_text = f"error: {result_text}"
         call_id_suffix = f" [call_id={content.call_id}]" if content.call_id else ""
         return f"mcp_tool_result: {result_text}{call_id_suffix}"
     if content.type in ("function_approval_request", "function_approval_response"):

python/packages/core/agent_framework/_tools.py

@@ -1462,7 +1462,8 @@ class FunctionInvocationConfiguration(TypedDict, total=False):
     - ``additional_tools``: Extra tools available during execution but not
       advertised to the model in the tool list.
     - ``include_detailed_errors``: Whether to include exception details in the
-      function result returned to the model.
+      function result returned to the model. Exception text may contain sensitive
+      information regardless of its source, so enable this only for a trusted channel.
 
     Note:
         ``max_iterations``, ``max_function_calls``, and ``max_duration_seconds``

python/packages/core/agent_framework/_types.py

@@ -40,6 +40,8 @@
 
 logger = logging.getLogger("agent_framework")
 
+_SERIALIZED_EXCEPTION_MARKER: Final[str] = "FunctionInvocationError"
+
 if TYPE_CHECKING:
     from pydantic import BaseModel
 
@@ -272,16 +274,24 @@ def _validate_uri(uri: str, media_type: str | None) -> dict[str, Any]:
     raise ContentError("URI must contain a scheme (e.g., http://, data:, file://)")
 
 
-def _serialize_value(value: Any, exclude_none: bool) -> Any:
+def _serialize_value(value: Any, exclude_none: bool, *, redact_exception: bool = True) -> Any:
     """Recursively serialize a value for to_dict."""
     if value is None:
         return None
     if isinstance(value, Content):
-        return value.to_dict(exclude_none=exclude_none)
+        return value._to_dict(  # pyright: ignore[reportPrivateUsage]
+            exclude_none=exclude_none, redact_exception=redact_exception
+        )
     if isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray)):
-        return [_serialize_value(item, exclude_none) for item in cast(Iterable[Any], value)]
+        return [
+            _serialize_value(item, exclude_none, redact_exception=redact_exception)
+            for item in cast(Iterable[Any], value)
+        ]
     if isinstance(value, Mapping):
-        return {k: _serialize_value(v, exclude_none) for k, v in value.items()}  # type: ignore[reportUnknownVariableType]
+        return {
+            k: _serialize_value(v, exclude_none, redact_exception=redact_exception)
+            for k, v in cast(Mapping[Any, Any], value).items()
+        }
     if hasattr(value, "to_dict"):
         return value.to_dict()  # type: ignore[call-arg]
     return value
@@ -480,6 +490,10 @@ class Content:
     This class provides a single unified type that handles all content variants.
     Use the class methods like `Content.from_text()`, `Content.from_data()`,
     `Content.from_uri()`, etc. to create instances.
+
+    The ``exception`` field is host-internal diagnostic state. Its value may originate from a tool, middleware,
+    provider, or caller and must always be treated as potentially sensitive. Dictionary serialization replaces it
+    with a fixed marker that preserves failure state; channel-visible error information belongs in the public result.
     """
 
     _SHALLOW_COPY_FIELDS: ClassVar[set[str]] = {"raw_representation"}
@@ -857,8 +871,8 @@ def from_function_call(
         Keyword Args:
             arguments: The arguments for the requested function call. May be a JSON string, a mapping that can be
                 serialized as arguments, or None when no arguments were provided.
-            exception: Error information associated with the function call, if the provider returned the call in an
-                error state.
+            exception: Host-internal diagnostic information when the provider returned the call in an error state.
+                Treat it as potentially sensitive regardless of its source; serialization replaces it with a marker.
             informational_only: Whether the function call is present only for transcript fidelity and should not be
                 executed by Agent Framework function invocation.
             id: Stable Agent Framework identity for this occurrence. When omitted, the function invocation layer
@@ -907,7 +921,9 @@ def from_function_result(
             result: The tool output.  Accepts a ``list[Content]`` (the canonical
                 form produced by :meth:`~FunctionTool.parse_result`), a plain
                 ``str``, or any other value (which is stringified).
-            exception: The exception message if the function call failed.
+            exception: Host-internal diagnostic information when the function call failed. Treat it as potentially
+                sensitive regardless of whether it came from a tool, middleware, provider, or caller. Serialization
+                replaces it with a fixed failure marker; use ``result`` for channel-visible error text.
             annotations: Optional annotations for the content.
             additional_properties: Optional additional properties.
             raw_representation: Optional raw representation from the provider.
@@ -1408,7 +1424,21 @@ def to_function_approval_response(
         )
 
     def to_dict(self, *, exclude_none: bool = True, exclude: set[str] | None = None) -> dict[str, Any]:
-        """Serialize the content to a dictionary."""
+        """Serialize content without host-internal exception diagnostics.
+
+        Exception diagnostics are replaced with a fixed marker regardless of their source because they may contain
+        sensitive information. The marker preserves failure status across persistence round-trips.
+        """
+        return self._to_dict(exclude_none=exclude_none, exclude=exclude, redact_exception=True)
+
+    def _to_dict(
+        self,
+        *,
+        exclude_none: bool,
+        exclude: set[str] | None = None,
+        redact_exception: bool,
+    ) -> dict[str, Any]:
+        """Serialize content with explicit control over internal exception redaction."""
         fields_to_capture = (
             "text",
             "protected_data",
@@ -1456,11 +1486,13 @@ def to_dict(self, *, exclude_none: bool = True, exclude: set[str] | None = None)
             value = getattr(self, field, None)
             if field in exclude:
                 continue
+            if field == "exception" and value is not None and redact_exception:
+                value = _SERIALIZED_EXCEPTION_MARKER
             if field == "informational_only" and (self.type != "function_call" or not value):
                 continue
             if exclude_none and value is None:
                 continue
-            result[field] = _serialize_value(value, exclude_none)
+            result[field] = _serialize_value(value, exclude_none, redact_exception=redact_exception)
 
         if "annotations" not in exclude and self.annotations is not None:
             result["annotations"] = [dict(annotation) for annotation in self.annotations]
@@ -1471,7 +1503,10 @@ def __eq__(self, other: object) -> bool:
         """Check if two Content instances are equal by comparing their dict representations."""
         if not isinstance(other, Content):
             return False
-        return self.to_dict(exclude_none=False) == other.to_dict(exclude_none=False)
+        return self._to_dict(exclude_none=False, redact_exception=False) == other._to_dict(
+            exclude_none=False,
+            redact_exception=False,
+        )
 
     def __str__(self) -> str:
         """Return a string representation of the Content."""