Repository: phidatahq/phidata
File: libs/agno/agno/os/interfaces/agui/utils.py (Python)

Code before:
    def set_pending_tool_calls_parent_id(self, parent_id: str) -> None:
        """Set the parent message ID for upcoming tool calls."""
        self.pending_tool_calls_parent_id = parent_id

    def clear_pending_tool_calls_parent_id(self) -> None:
        """Clear the pending parent ID when a new text message starts."""
        self.pending_tool_calls_parent_id = ""


def convert_agui_messages_to_agno_messages(messages: List[AGUIMessage]) -> List[Message]:
    """Convert AG-UI messages to Agno messages."""
    # First pass: collect all tool_call_ids that have results
    tool_call_ids_with_results: Set[str] = set()
    for msg in messages:
        if msg.role == "tool" and msg.tool_call_id:
            tool_call_ids_with_results.add(msg.tool_call_id)

    # Second pass: convert messages
    result: List[Message] = []
    seen_tool_call_ids: Set[str] = set()

    for msg in messages:
        if msg.role == "tool":
            # Deduplicate tool results - keep only first occurrence
            if msg.tool_call_id in seen_tool_call_ids:
                log_warning(f"Skipping duplicate AGUI tool result: {msg.tool_call_id}")
                continue
            seen_tool_call_ids.add(msg.tool_call_id)
            result.append(Message(role="tool", tool_call_id=msg.tool_call_id, content=msg.content))

        elif msg.role == "assistant":
            tool_calls = None
            if msg.tool_calls:
                # Filter tool_calls to only those with results in this message sequence
                filtered_calls = [call for call in msg.tool_calls if call.id in tool_call_ids_with_results]
                if filtered_calls:
                    tool_calls = [call.model_dump() for call in filtered_calls]
            result.append(Message(role="assistant", content=msg.content, tool_calls=tool_calls))

        elif msg.role == "user":
            result.append(Message(role="user", content=msg.content))

        elif msg.role == "system":
            pass  # Skip - agent builds its own system message from configuration

        else:
            log_warning(f"Unknown AGUI message role: {msg.role}")

    return result



Review comment (bug):
```suggestion
                log_debug(f"Skipping duplicate AGUI tool result: {msg.tool_call_id}")
```

Code after:
    def set_pending_tool_calls_parent_id(self, parent_id: str) -> None:
        """Set the parent message ID for upcoming tool calls."""
        self.pending_tool_calls_parent_id = parent_id

    def clear_pending_tool_calls_parent_id(self) -> None:
        """Clear the pending parent ID when a new text message starts."""
        self.pending_tool_calls_parent_id = ""


def convert_agui_messages_to_agno_messages(messages: List[AGUIMessage]) -> List[Message]:
    """Convert AG-UI messages to Agno messages."""
    # First pass: collect all tool_call_ids that have results
    tool_call_ids_with_results: Set[str] = set()
    for msg in messages:
        if msg.role == "tool" and msg.tool_call_id:
            tool_call_ids_with_results.add(msg.tool_call_id)

    # Second pass: convert messages
    result: List[Message] = []
    seen_tool_call_ids: Set[str] = set()

    for msg in messages:
        if msg.role == "tool":
            # Deduplicate tool results - keep only first occurrence
            if msg.tool_call_id in seen_tool_call_ids:
                log_debug(f"Skipping duplicate AGUI tool result: {msg.tool_call_id}")
                continue
            seen_tool_call_ids.add(msg.tool_call_id)
            result.append(Message(role="tool", tool_call_id=msg.tool_call_id, content=msg.content))

        elif msg.role == "assistant":
            tool_calls = None
            if msg.tool_calls:
                # Filter tool_calls to only those with results in this message sequence
                filtered_calls = [call for call in msg.tool_calls if call.id in tool_call_ids_with_results]
                if filtered_calls:
                    tool_calls = [call.model_dump() for call in filtered_calls]
            result.append(Message(role="assistant", content=msg.content, tool_calls=tool_calls))

        elif msg.role == "user":
            result.append(Message(role="user", content=msg.content))

        elif msg.role == "system":
            pass  # Skip - agent builds its own system message from configuration

        else:
            log_warning(f"Unknown AGUI message role: {msg.role}")

    return result