Python: fix(orchestrations): preserve multimodal content during agent handoff

python/packages/orchestrations/agent_framework_orchestrations/_orchestrator_helpers.py

@@ -28,22 +28,41 @@ def clean_conversation_for_handoff(conversation: list[Message]) -> list[Message]
     Args:
         conversation: Full conversation history, including tool-control content
     Returns:
-        Cleaned conversation history with only text content, suitable for handoff routing
+        Cleaned conversation history with semantic multimodal content preserved, suitable for handoff routing
     """
+    ALLOWED_CONTENT_TYPES = {
+        "text",
+        "data",
+        "uri",
+        "hosted_file",
+        "hosted_vector_store",
+    }
+
     cleaned: list[Message] = []
     for msg in conversation:
-        # Keep only plain text history for handoff routing. Tool-control content
+        # Keep non-tool history for handoff routing. Tool-control content
         # (function_call/function_result/approval payloads) is runtime-only and
         # must not be replayed in future model turns.
-        text_parts = [content.text for content in msg.contents if content.type == "text" and content.text]
-        # TODO(@taochen): This is a simplified check that considers any non-text content as a tool call.
-        # We need to enhance this logic to specifically identify tool related contents.
-        if not text_parts:
+        retained_contents = []
+        for content in msg.contents:
+            ctype = getattr(content, "type", "text")
+
+            # Skip disallowed types (tools, usage, errors, etc.)
+            if ctype not in ALLOWED_CONTENT_TYPES:
+                continue
+
+            # Skip empty text parts
+            if ctype == "text" and not getattr(content, "text", None):
+                continue
+
+            retained_contents.append(content)
+
+        if not retained_contents:
             continue
 
         msg_copy = Message(
             role=msg.role,
-            contents=[" ".join(text_parts)],
+            contents=retained_contents,
             author_name=msg.author_name,
             additional_properties=dict(msg.additional_properties) if msg.additional_properties else None,
         )

python/packages/orchestrations/tests/test_handoff.py

@@ -985,8 +985,8 @@ async def observe_properties(context: AgentContext, call_next):
     assert cloned_additional_properties is not coordinator.additional_properties
 
 
-def test_clean_conversation_for_handoff_keeps_text_only_history() -> None:
-    """Tool-control messages must be excluded from persisted handoff history."""
+def test_clean_conversation_for_handoff_keeps_allowlist_history() -> None:
+    """Tool-control messages must be excluded, but text and multimodal data must be preserved."""
     function_call = Content.from_function_call(
         call_id="handoff-call-1",
         name="handoff_to_refund_agent",
@@ -997,9 +997,12 @@ def test_clean_conversation_for_handoff_keeps_text_only_history() -> None:
         id="approval-1",
         function_call=function_call,
     )
+
+    # Simulate a user attaching an image to their message
+    multimodal_content = Content(type="uri", uri="https://example.com/image.png", media_type="image/png")
 
     conversation = [
-        Message(role="user", contents=["My order arrived damaged."]),
+        Message(role="user", contents=["My order arrived damaged.", multimodal_content]),
         Message(
             role="assistant",
             contents=[
@@ -1017,10 +1020,24 @@ def test_clean_conversation_for_handoff_keeps_text_only_history() -> None:
 
     cleaned = clean_conversation_for_handoff(conversation)
     assert [message.role for message in cleaned] == ["user", "assistant"]
+
+    # Assert Text is preserved
     assert [message.text for message in cleaned] == [
         "My order arrived damaged.",
         "Triage Agent: Routing you to Refund.",
     ]
+
+    # Assert Multimodal URI is preserved in the first user message
+    user_contents = cleaned[0].contents
+    assert len(user_contents) == 2
+    assert user_contents[0].type == "text"
+    assert user_contents[1].type == "uri"
+    assert user_contents[1].uri == "https://example.com/image.png"
+    assert getattr(user_contents[1], "media_type", None) == "image/png"
+
+    # Assert Tool call is stripped from the assistant message
+    assistant_contents = [c.type for c in cleaned[1].contents]
+    assert "function_call" not in assistant_contents
 
 
 async def test_autonomous_mode_yields_output_without_user_request():

python/packages/orchestrations/tests/test_sequential.py

@@ -246,6 +246,35 @@ async def test_sequential_checkpoint_resume_round_trip() -> None:
     assert baseline_text == resumed_text
 
 
+async def test_sequential_builder_preserves_multimodal_content() -> None:
+    """Ensure that multimodal content (like URI) is preserved and passed down to subsequent agents in the sequence."""
+    class _InspectorAgent(BaseAgent):
+        def run(
+            self, messages: AgentRunInputs | None = None, **kwargs: Any
+        ) -> Awaitable[AgentResponse[Any]] | ResponseStream[AgentResponseUpdate, AgentResponse[Any]]:
+            async def _run() -> AgentResponse:
+                uri_contents = [
+                    c.uri for m in (messages or []) for c in getattr(m, "contents", []) if getattr(c, "type", "") == "uri"
+                ]
+                return AgentResponse(messages=[Message("assistant", [f"Found URIs: {uri_contents}"])])
+            return _run()
+
+    a1 = _EchoAgent(id="agent1", name="A1")
+    inspector = _InspectorAgent(id="inspector", name="Inspector")
+
+    wf = SequentialBuilder(participants=[a1, inspector]).build()
+
+    multimodal_msg = Message("user", [Content(type="text", text="Look at this"), Content(type="uri", uri="https://example.com/image.png", media_type="image/png")])
+    output_events = [ev for ev in await wf.run([multimodal_msg]) if ev.type == "output"]
+
+    assert len(output_events) == 1
+    response = output_events[0].data
+    assert isinstance(response, AgentResponse)
+
+    combined = " ".join(m.text for m in response.messages)
+    assert "https://example.com/image.png" in combined
+
+
 async def test_sequential_checkpoint_runtime_only() -> None:
     """Test checkpointing configured ONLY at runtime, not at build time."""
     storage = InMemoryCheckpointStorage()