Python: Enforce FIDES MCP IFC labels

python/packages/core/agent_framework/_mcp.py

@@ -205,6 +205,7 @@ class MCPSpecificApproval(TypedDict, total=False):
 _MCP_IS_TOOL_KEY = "_mcp_is_tool"
 _MCP_NORMALIZED_NAME_KEY = "_mcp_normalized_name"
 _MCP_TOOL_RESULT_HOST_PAYLOAD_KEY = "_mcp_tool_result_host_payload"
+_MCP_TOOL_RESULT_SECURITY_META_KEY = "_mcp_tool_result_security_meta"
 _MCP_PROGRESSIVE_LIST_TOOL_NAME = "list_mcp_tools"
 _MCP_PROGRESSIVE_LOAD_TOOL_NAME = "load_tool"
 _MCP_PROGRESSIVE_UNLOAD_TOOL_NAME = "unload_tool"
@@ -251,6 +252,7 @@ class MCPSpecificApproval(TypedDict, total=False):
 _MCP_SUBSCRIPTION_RETRY_INITIAL_DELAY = 0.1
 _MCP_SUBSCRIPTION_RETRY_MAX_DELAY = 2.0
 _DEFAULT_MCP_HOST_PAYLOAD_SIZE_BYTES = 1024 * 1024
+_MAX_MCP_SECURITY_META_SIZE_BYTES = 4096
 
 
 @dataclass
@@ -262,8 +264,45 @@ class _MCPHostPayloadCapture:
     apply_server_meta_to_model_items: bool
     host_payload: dict[str, Any] | None = None
     meta: dict[str, Any] | None = None
+    security_meta: dict[str, Any] | None = None
     recorded: bool = False
     meta_prepared: bool = False
+    security_meta_prepared: bool = False
+
+    def prepare_security_meta(self, mcp_type: Any) -> dict[str, Any] | None:
+        """Retain security-relevant MCP metadata outside generic payload budgets."""
+        if self.security_meta_prepared:
+            return self.security_meta
+        self.security_meta_prepared = True
+
+        raw_meta = getattr(mcp_type, "meta", None)
+        if not isinstance(raw_meta, Mapping) or "ifc" not in raw_meta:
+            return None
+
+        raw_meta_mapping = cast(Mapping[str, Any], raw_meta)
+        raw_security_meta: dict[str, Any] = {"ifc": raw_meta_mapping["ifc"]}
+        if _json_size_exceeds(raw_security_meta, _MAX_MCP_SECURITY_META_SIZE_BYTES):
+            logger.warning("MCP result _meta.ifc exceeds the independent security metadata limit.")
+            self.security_meta = {"retained": False}
+            return self.security_meta
+
+        try:
+            copied = cast(
+                dict[str, Any],
+                to_jsonable_python(raw_security_meta, fallback=str),
+            )
+        except (TypeError, ValueError):
+            logger.warning("MCP result _meta.ifc could not be retained safely.")
+            self.security_meta = {"retained": False}
+            return self.security_meta
+
+        if len(json.dumps(copied).encode("utf-8")) > _MAX_MCP_SECURITY_META_SIZE_BYTES:
+            logger.warning("MCP result _meta.ifc exceeds the independent security metadata limit.")
+            self.security_meta = {"retained": False}
+            return self.security_meta
+
+        self.security_meta = {"retained": True, "meta": copied}
+        return self.security_meta
 
     def prepare_meta(self, mcp_type: Any) -> dict[str, Any] | None:
         if self.meta_prepared:
@@ -293,6 +332,7 @@ def record(self, mcp_type: Any) -> None:
         if self.recorded:
             return
         self.recorded = True
+        self.prepare_security_meta(mcp_type)
         self.prepare_meta(mcp_type)
 
         effective_limit = self.max_size_bytes
@@ -325,8 +365,16 @@ def to_carrier(self) -> _FunctionResultCarrier:
         return _FunctionResultCarrier(
             additional_properties=additional_properties,
             item_additional_properties=item_additional_properties,
+            private_metadata=(
+                {_MCP_TOOL_RESULT_SECURITY_META_KEY: self.security_meta} if self.security_meta is not None else {}
+            ),
             exclusive_outer_keys=frozenset({_MCP_TOOL_RESULT_HOST_PAYLOAD_KEY}),
-            exclusive_item_keys=frozenset({"_meta"}),
+            exclusive_item_keys=frozenset({
+                "_meta",
+                "_security_label_authoritative",
+                "_security_label_authoritative_confidentiality",
+                _MCP_TOOL_RESULT_SECURITY_META_KEY,
+            }),
             result_already_parsed=True,
         )
 
@@ -338,8 +386,11 @@ def prepare_model_result(self, parsed: str | list[Content]) -> list[Content]:
             updated_item.additional_properties = dict(updated_item.additional_properties)
             updated_item.additional_properties.pop(_MCP_TOOL_RESULT_HOST_PAYLOAD_KEY, None)
             updated_item.additional_properties.pop("_meta", None)
+            updated_item.additional_properties.pop(_MCP_TOOL_RESULT_SECURITY_META_KEY, None)
             if self.apply_server_meta_to_model_items and self.meta is not None:
                 updated_item.additional_properties["_meta"] = self.meta
+            if self.security_meta is not None:
+                updated_item.additional_properties[_MCP_TOOL_RESULT_SECURITY_META_KEY] = self.security_meta
             items[index] = updated_item
         return items
 
@@ -1413,6 +1464,7 @@ def _parse_tool_result_from_mcp(
 
         capture = _mcp_host_payload_capture.get()
         if capture is not None:
+            capture.prepare_security_meta(mcp_type)
             meta = capture.prepare_meta(mcp_type)
         else:
             raw_meta = mcp_type.meta

python/packages/core/agent_framework/_tools.py

@@ -22,7 +22,7 @@
     Sequence,
 )
 from contextlib import suppress
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from datetime import date, datetime, timedelta, timezone, tzinfo
 from datetime import time as datetime_time
 from decimal import Decimal
@@ -312,6 +312,10 @@ class _FunctionResultCarrier:
 
     additional_properties: dict[str, Any]
     item_additional_properties: dict[str, Any]
+    private_metadata: dict[str, Any] = field(default_factory=dict[str, Any])
+    parsed_result_postprocessor: Callable[[list[Content]], list[Content]] | None = None
+    parser_exception_handler: Callable[[], None] | None = None
+    error_result_finalizer: Callable[[], None] | None = None
     exclusive_outer_keys: frozenset[str] = frozenset()
     exclusive_item_keys: frozenset[str] = frozenset()
     result_already_parsed: bool = False
@@ -1218,7 +1222,13 @@ async def invoke(
             elif callable(configured_parser):
                 try:
                     parsed = configured_parser(result)
+                    if isinstance(parsed, str):
+                        parsed = [Content.from_text(parsed)]
+                    if isinstance(carrier, _FunctionResultCarrier) and carrier.parsed_result_postprocessor is not None:
+                        parsed = carrier.parsed_result_postprocessor(parsed)
                 except Exception as exception:
+                    if isinstance(carrier, _FunctionResultCarrier) and carrier.parser_exception_handler is not None:
+                        carrier.parser_exception_handler()
                     self.invocation_exception_count += 1
                     if OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED:
                         logger.error(f"Function {self.name}: result parser failed. Error: {exception}")
@@ -1312,7 +1322,16 @@ async def invoke(
                 elif callable(configured_parser):
                     try:
                         parsed = configured_parser(result)
+                        if isinstance(parsed, str):
+                            parsed = [Content.from_text(parsed)]
+                        if (
+                            isinstance(carrier, _FunctionResultCarrier)
+                            and carrier.parsed_result_postprocessor is not None
+                        ):
+                            parsed = carrier.parsed_result_postprocessor(parsed)
                     except Exception as exception:
+                        if isinstance(carrier, _FunctionResultCarrier) and carrier.parser_exception_handler is not None:
+                            carrier.parser_exception_handler()
                         self.invocation_exception_count += 1
                         metric_attributes[OtelAttr.ERROR_TYPE] = type(exception).__name__
                         if OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED:
@@ -2164,21 +2183,26 @@ def _finalize_function_result(
         exception=exception,
         additional_properties=additional_properties,
     )
-    if carrier is None or function_result.items is None:
-        return function_result
-
-    updated_items = list(function_result.items)
-    for index, item in enumerate(updated_items):
-        updated_item = copy.copy(item)
-        updated_item.additional_properties = dict(updated_item.additional_properties)
-        for key in carrier.exclusive_outer_keys | carrier.exclusive_item_keys:
-            updated_item.additional_properties.pop(key, None)
-        updated_item.additional_properties = merge_additional_properties(
-            updated_item.additional_properties,
-            carrier.item_additional_properties,
-        )
-        updated_items[index] = updated_item
-    function_result.items = updated_items
+    if carrier is not None and function_result.items is not None:
+        updated_items = list(function_result.items)
+        for index, item in enumerate(updated_items):
+            updated_item = copy.copy(item)
+            updated_item.additional_properties = dict(updated_item.additional_properties)
+            for key in carrier.exclusive_outer_keys | carrier.exclusive_item_keys:
+                updated_item.additional_properties.pop(key, None)
+            is_hidden_variable_reference = (
+                updated_item.type == "text" and updated_item.additional_properties.get("_variable_reference") is True
+            )
+            if not is_hidden_variable_reference:
+                updated_item.additional_properties = merge_additional_properties(
+                    updated_item.additional_properties,
+                    carrier.item_additional_properties,
+                )
+            updated_items[index] = updated_item
+        function_result.items = updated_items
+
+    if exception is not None and carrier is not None and carrier.error_result_finalizer is not None:
+        carrier.error_result_finalizer()
     return function_result