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

python/packages/orchestrations/agent_framework_orchestrations/_orchestrator_helpers.py

@@ -8,42 +8,66 @@
 
 import logging
 
-from agent_framework._types import Message
+from agent_framework._types import Content, Message
 
 logger = logging.getLogger(__name__)
 
 
 def clean_conversation_for_handoff(conversation: list[Message]) -> list[Message]:
-    """Keep only plain text chat history for handoff routing.
+    """Clean conversation history for handoff routing.
 
     Handoff executors must not replay prior tool-control artifacts (function calls,
     tool outputs, approval payloads) into future model turns, or providers may reject
     the next request due to unmatched tool-call state.
 
-    This helper builds a text-only copy of the conversation:
-    - Drops all non-text content from every message.
-    - Drops messages with no remaining text content.
-    - Preserves original roles and author names for retained text messages.
+    This helper preserves semantic content:
+    - For `user` messages, preserves text and multimodal content (data, uri, hosted_file, hosted_vector_store).
+    - For non-user messages (assistant, system, etc.), preserves only text content to avoid serializing input-only
+      multimodal parts into assistant roles on model providers.
+    - Drops tool-control payloads (function_call, function_result, approval payloads, etc.).
+    - Drops messages with no remaining content.
+    - Preserves original roles, author names, and additional properties for retained messages.
 
     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 for user messages,
+        suitable for handoff routing.
     """
+    USER_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
-        # (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:
+        is_user = msg.role == "user" or str(msg.role).lower() == "user"
+        allowed_types = USER_ALLOWED_CONTENT_TYPES if is_user else {"text"}
+
+        retained_contents: list[Content] = []
+        for content in msg.contents:
+            ctype = getattr(content, "type", "text")
+
+            # Skip disallowed types
+            if ctype not in allowed_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, multimodal data preserved for user, and text-only for assistant."""
     function_call = Content.from_function_call(
         call_id="handoff-call-1",
         name="handoff_to_refund_agent",
@@ -998,13 +998,32 @@ def test_clean_conversation_for_handoff_keeps_text_only_history() -> None:
         function_call=function_call,
     )
 
+    # Simulate a user attaching multiple multimodal types to their message
+    uri_content = Content(type="uri", uri="https://example.com/image.png", media_type="image/png")
+    data_content = Content.from_data(data=b"fake-bytes", media_type="image/jpeg")
+    file_content = Content(type="hosted_file", file_id="file-123")
+    vector_content = Content(type="hosted_vector_store", vector_store_id="vs-456")
+
+    # Simulate an assistant containing uri/data that should not be replayed as input-only items
+    assistant_multimodal_content = Content(type="uri", uri="https://example.com/output.png", media_type="image/png")
+
     conversation = [
-        Message(role="user", contents=["My order arrived damaged."]),
+        Message(
+            role="user",
+            contents=[
+                "My order arrived damaged.",
+                uri_content,
+                data_content,
+                file_content,
+                vector_content,
+            ],
+        ),
         Message(
             role="assistant",
             contents=[
                 function_call,
                 Content.from_text(text="Triage Agent: Routing you to Refund."),
+                assistant_multimodal_content,
             ],
         ),
         Message(role="tool", contents=[Content.from_function_result(call_id="handoff-call-1", result="ok")]),
@@ -1017,11 +1036,26 @@ 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 all Multimodal contents are preserved in the user message
+    user_contents = cleaned[0].contents
+    assert [c.type for c in user_contents] == ["text", "uri", "data", "hosted_file", "hosted_vector_store"]
+    assert user_contents[1].uri == "https://example.com/image.png"
+    assert user_contents[2].type == "data"
+    assert user_contents[2].uri is not None and user_contents[2].uri.startswith("data:image/jpeg;base64,")
+    assert getattr(user_contents[3], "file_id", None) == "file-123"
+    assert getattr(user_contents[4], "vector_store_id", None) == "vs-456"
+
+    # Assert Tool call and assistant multimodal contents are stripped from the assistant message
+    assistant_contents = [c.type for c in cleaned[1].contents]
+    assert assistant_contents == ["text"]
+
 
 async def test_autonomous_mode_yields_output_without_user_request():
     """Ensure autonomous interaction mode yields output without requesting user input."""

python/packages/orchestrations/tests/test_sequential.py

@@ -246,6 +246,44 @@ 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:
+                assert isinstance(messages, Sequence) and not isinstance(messages, str)
+                uri_contents = [
+                    c.uri for m in messages 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()