diff --git a/homeassistant/components/mpd/media_player.py b/homeassistant/components/mpd/media_player.py
index 8eab83b5d41..9b3adb38e0c 100644
--- a/homeassistant/components/mpd/media_player.py
+++ b/homeassistant/components/mpd/media_player.py
@@ -1,11 +1,13 @@
 """Support to interact with a Music Player Daemon."""
 from __future__ import annotations
 
-from contextlib import suppress
+import asyncio
+from contextlib import asynccontextmanager, suppress
 from datetime import timedelta
 import hashlib
 import logging
 import os
+from socket import gaierror
 from typing import Any
 
 import mpd
@@ -92,11 +94,11 @@ class MpdDevice(MediaPlayerEntity):
         self._name = name
         self.password = password
 
-        self._status = None
+        self._status = {}
         self._currentsong = None
         self._playlists = None
         self._currentplaylist = None
-        self._is_connected = False
+        self._is_available = None
         self._muted = False
         self._muted_volume = None
         self._media_position_updated_at = None
@@ -104,67 +106,88 @@ class MpdDevice(MediaPlayerEntity):
         self._media_image_hash = None
         # Track if the song changed so image doesn't have to be loaded every update.
         self._media_image_file = None
-        self._commands = None
 
         # set up MPD client
         self._client = MPDClient()
         self._client.timeout = 30
-        self._client.idletimeout = None
+        self._client.idletimeout = 10
+        self._client_lock = asyncio.Lock()
 
-    async def _connect(self):
-        """Connect to MPD."""
-        try:
-            await self._client.connect(self.server, self.port)
-
-            if self.password is not None:
-                await self._client.password(self.password)
-        except mpd.ConnectionError:
-            return
-
-        self._is_connected = True
-
-    def _disconnect(self):
-        """Disconnect from MPD."""
-        with suppress(mpd.ConnectionError):
-            self._client.disconnect()
-        self._is_connected = False
-        self._status = None
-
-    async def _fetch_status(self):
-        """Fetch status from MPD."""
-        self._status = await self._client.status()
-        self._currentsong = await self._client.currentsong()
-        await self._async_update_media_image_hash()
-
-        if (position := self._status.get("elapsed")) is None:
-            position = self._status.get("time")
-
-            if isinstance(position, str) and ":" in position:
-                position = position.split(":")[0]
-
-        if position is not None and self._media_position != position:
-            self._media_position_updated_at = dt_util.utcnow()
-            self._media_position = int(float(position))
-
-        await self._update_playlists()
-
-    @property
-    def available(self):
-        """Return true if MPD is available and connected."""
-        return self._is_connected
+    # Instead of relying on python-mpd2 to maintain a (persistent) connection to
+    # MPD, the below explicitly sets up a *non*-persistent connection. This is
+    # done to workaround the issue as described in:
+    #   <https://github.com/Mic92/python-mpd2/issues/31>
+    @asynccontextmanager
+    async def connection(self):
+        """Handle MPD connect and disconnect."""
+        async with self._client_lock:
+            try:
+                # MPDClient.connect() doesn't always respect its timeout. To
+                # prevent a deadlock, enforce an additional (slightly longer)
+                # timeout on the coroutine itself.
+                try:
+                    async with asyncio.timeout(self._client.timeout + 5):
+                        await self._client.connect(self.server, self.port)
+                except asyncio.TimeoutError as error:
+                    # TimeoutError has no message (which hinders logging further
+                    # down the line), so provide one.
+                    raise asyncio.TimeoutError(
+                        "Connection attempt timed out"
+                    ) from error
+                if self.password is not None:
+                    await self._client.password(self.password)
+                self._is_available = True
+                yield
+            except (
+                asyncio.TimeoutError,
+                gaierror,
+                mpd.ConnectionError,
+                OSError,
+            ) as error:
+                # Log a warning during startup or when previously connected; for
+                # subsequent errors a debug message is sufficient.
+                log_level = logging.DEBUG
+                if self._is_available is not False:
+                    log_level = logging.WARNING
+                _LOGGER.log(
+                    log_level, "Error connecting to '%s': %s", self.server, error
+                )
+                self._is_available = False
+                self._status = {}
+                # Also yield on failure. Handling mpd.ConnectionErrors caused by
+                # attempting to control a disconnected client is the
+                # responsibility of the caller.
+                yield
+            finally:
+                with suppress(mpd.ConnectionError):
+                    self._client.disconnect()
 
     async def async_update(self) -> None:
-        """Get the latest data and update the state."""
-        try:
-            if not self._is_connected:
-                await self._connect()
-                self._commands = list(await self._client.commands())
+        """Get the latest data from MPD and update the state."""
+        async with self.connection():
+            try:
+                self._status = await self._client.status()
+                self._currentsong = await self._client.currentsong()
+                await self._async_update_media_image_hash()
 
-            await self._fetch_status()
-        except (mpd.ConnectionError, OSError, ValueError) as error:
-            # Cleanly disconnect in case connection is not in valid state
-            _LOGGER.debug("Error updating status: %s", error)
-            self._disconnect()
+                if (position := self._status.get("elapsed")) is None:
+                    position = self._status.get("time")
+
+                    if isinstance(position, str) and ":" in position:
+                        position = position.split(":")[0]
+
+                if position is not None and self._media_position != position:
+                    self._media_position_updated_at = dt_util.utcnow()
+                    self._media_position = int(float(position))
+
+                await self._update_playlists()
+            except (mpd.ConnectionError, ValueError) as error:
+                _LOGGER.debug("Error updating status: %s", error)
+
+    @property
+    def available(self) -> bool:
+        """Return true if MPD is available and connected."""
+        return self._is_available is True
 
     @property
     def name(self):
@@ -174,13 +197,13 @@ class MpdDevice(MediaPlayerEntity):
     @property
     def state(self) -> MediaPlayerState:
         """Return the media state."""
-        if self._status is None:
+        if not self._status:
             return MediaPlayerState.OFF
-        if self._status["state"] == "play":
+        if self._status.get("state") == "play":
             return MediaPlayerState.PLAYING
-        if self._status["state"] == "pause":
+        if self._status.get("state") == "pause":
             return MediaPlayerState.PAUSED
-        if self._status["state"] == "stop":
+        if self._status.get("state") == "stop":
             return MediaPlayerState.OFF
 
         return MediaPlayerState.OFF
@@ -259,20 +282,26 @@ class MpdDevice(MediaPlayerEntity):
 
     async def async_get_media_image(self) -> tuple[bytes | None, str | None]:
         """Fetch media image of current playing track."""
-        if not (file := self._currentsong.get("file")):
-            return None, None
-        response = await self._async_get_file_image_response(file)
-        if response is None:
-            return None, None
+        async with self.connection():
+            if self._currentsong is None or not (file := self._currentsong.get("file")):
+                return None, None
 
-        image = bytes(response["binary"])
-        mime = response.get(
-            "type", "image/png"
-        )  # readpicture has type, albumart does not
-        return (image, mime)
+            with suppress(mpd.ConnectionError):
+                response = await self._async_get_file_image_response(file)
+            if response is None:
+                return None, None
+
+            image = bytes(response["binary"])
+            mime = response.get(
+                "type", "image/png"
+            )  # readpicture has type, albumart does not
+            return (image, mime)
 
     async def _async_update_media_image_hash(self):
         """Update the hash value for the media image."""
+        if self._currentsong is None:
+            return
+
         file = self._currentsong.get("file")
 
         if file == self._media_image_file:
@@ -295,16 +324,21 @@ class MpdDevice(MediaPlayerEntity):
         self._media_image_file = file
 
     async def _async_get_file_image_response(self, file):
-        # not all MPD implementations and versions support the `albumart` and `fetchpicture` commands
-        can_albumart = "albumart" in self._commands
-        can_readpicture = "readpicture" in self._commands
+        # not all MPD implementations and versions support the `albumart` and
+        # `fetchpicture` commands.
+        commands = []
+        with suppress(mpd.ConnectionError):
+            commands = list(await self._client.commands())
+        can_albumart = "albumart" in commands
+        can_readpicture = "readpicture" in commands
 
         response = None
 
         # read artwork embedded into the media file
         if can_readpicture:
             try:
-                response = await self._client.readpicture(file)
+                with suppress(mpd.ConnectionError):
+                    response = await self._client.readpicture(file)
             except mpd.CommandError as error:
                 if error.errno is not mpd.FailureResponseCode.NO_EXIST:
                     _LOGGER.warning(
@@ -315,7 +349,8 @@ class MpdDevice(MediaPlayerEntity):
         # read artwork contained in the media directory (cover.{jpg,png,tiff,bmp}) if none is embedded
         if can_albumart and not response:
             try:
-                response = await self._client.albumart(file)
+                with suppress(mpd.ConnectionError):
+                    response = await self._client.albumart(file)
             except mpd.CommandError as error:
                 if error.errno is not mpd.FailureResponseCode.NO_EXIST:
                     _LOGGER.warning(
@@ -339,7 +374,7 @@ class MpdDevice(MediaPlayerEntity):
     @property
     def supported_features(self) -> MediaPlayerEntityFeature:
         """Flag media player features that are supported."""
-        if self._status is None:
+        if not self._status:
             return MediaPlayerEntityFeature(0)
 
         supported = SUPPORT_MPD
@@ -373,55 +408,64 @@ class MpdDevice(MediaPlayerEntity):
         """Update available MPD playlists."""
         try:
             self._playlists = []
-            for playlist_data in await self._client.listplaylists():
-                self._playlists.append(playlist_data["playlist"])
+            with suppress(mpd.ConnectionError):
+                for playlist_data in await self._client.listplaylists():
+                    self._playlists.append(playlist_data["playlist"])
         except mpd.CommandError as error:
             self._playlists = None
             _LOGGER.warning("Playlists could not be updated: %s:", error)
 
     async def async_set_volume_level(self, volume: float) -> None:
         """Set volume of media player."""
-        if "volume" in self._status:
-            await self._client.setvol(int(volume * 100))
+        async with self.connection():
+            if "volume" in self._status:
+                await self._client.setvol(int(volume * 100))
 
     async def async_volume_up(self) -> None:
         """Service to send the MPD the command for volume up."""
-        if "volume" in self._status:
-            current_volume = int(self._status["volume"])
+        async with self.connection():
+            if "volume" in self._status:
+                current_volume = int(self._status["volume"])
 
-            if current_volume <= 100:
-                self._client.setvol(current_volume + 5)
+                if current_volume <= 100:
+                    self._client.setvol(current_volume + 5)
 
     async def async_volume_down(self) -> None:
         """Service to send the MPD the command for volume down."""
-        if "volume" in self._status:
-            current_volume = int(self._status["volume"])
+        async with self.connection():
+            if "volume" in self._status:
+                current_volume = int(self._status["volume"])
 
-            if current_volume >= 0:
-                await self._client.setvol(current_volume - 5)
+                if current_volume >= 0:
+                    await self._client.setvol(current_volume - 5)
 
     async def async_media_play(self) -> None:
         """Service to send the MPD the command for play/pause."""
-        if self._status["state"] == "pause":
-            await self._client.pause(0)
-        else:
-            await self._client.play()
+        async with self.connection():
+            if self._status.get("state") == "pause":
+                await self._client.pause(0)
+            else:
+                await self._client.play()
 
     async def async_media_pause(self) -> None:
         """Service to send the MPD the command for play/pause."""
-        await self._client.pause(1)
+        async with self.connection():
+            await self._client.pause(1)
 
     async def async_media_stop(self) -> None:
         """Service to send the MPD the command for stop."""
-        await self._client.stop()
+        async with self.connection():
+            await self._client.stop()
 
     async def async_media_next_track(self) -> None:
         """Service to send the MPD the command for next track."""
-        await self._client.next()
+        async with self.connection():
+            await self._client.next()
 
     async def async_media_previous_track(self) -> None:
         """Service to send the MPD the command for previous track."""
-        await self._client.previous()
+        async with self.connection():
+            await self._client.previous()
 
     async def async_mute_volume(self, mute: bool) -> None:
         """Mute. Emulated with set_volume_level."""
@@ -437,75 +481,82 @@ class MpdDevice(MediaPlayerEntity):
         self, media_type: MediaType | str, media_id: str, **kwargs: Any
     ) -> None:
         """Send the media player the command for playing a playlist."""
-        if media_source.is_media_source_id(media_id):
-            media_type = MediaType.MUSIC
-            play_item = await media_source.async_resolve_media(
-                self.hass, media_id, self.entity_id
-            )
-            media_id = async_process_play_media_url(self.hass, play_item.url)
+        async with self.connection():
+            if media_source.is_media_source_id(media_id):
+                media_type = MediaType.MUSIC
+                play_item = await media_source.async_resolve_media(
+                    self.hass, media_id, self.entity_id
+                )
+                media_id = async_process_play_media_url(self.hass, play_item.url)
 
-        if media_type == MediaType.PLAYLIST:
-            _LOGGER.debug("Playing playlist: %s", media_id)
-            if media_id in self._playlists:
-                self._currentplaylist = media_id
+            if media_type == MediaType.PLAYLIST:
+                _LOGGER.debug("Playing playlist: %s", media_id)
+                if media_id in self._playlists:
+                    self._currentplaylist = media_id
+                else:
+                    self._currentplaylist = None
+                    _LOGGER.warning("Unknown playlist name %s", media_id)
+                await self._client.clear()
+                await self._client.load(media_id)
+                await self._client.play()
             else:
+                await self._client.clear()
                 self._currentplaylist = None
-                _LOGGER.warning("Unknown playlist name %s", media_id)
-            await self._client.clear()
-            await self._client.load(media_id)
-            await self._client.play()
-        else:
-            await self._client.clear()
-            self._currentplaylist = None
-            await self._client.add(media_id)
-            await self._client.play()
+                await self._client.add(media_id)
+                await self._client.play()
 
     @property
     def repeat(self) -> RepeatMode:
         """Return current repeat mode."""
-        if self._status["repeat"] == "1":
-            if self._status["single"] == "1":
+        if self._status.get("repeat") == "1":
+            if self._status.get("single") == "1":
                 return RepeatMode.ONE
             return RepeatMode.ALL
         return RepeatMode.OFF
 
     async def async_set_repeat(self, repeat: RepeatMode) -> None:
         """Set repeat mode."""
-        if repeat == RepeatMode.OFF:
-            await self._client.repeat(0)
-            await self._client.single(0)
-        else:
-            await self._client.repeat(1)
-            if repeat == RepeatMode.ONE:
-                await self._client.single(1)
-            else:
+        async with self.connection():
+            if repeat == RepeatMode.OFF:
+                await self._client.repeat(0)
                 await self._client.single(0)
+            else:
+                await self._client.repeat(1)
+                if repeat == RepeatMode.ONE:
+                    await self._client.single(1)
+                else:
+                    await self._client.single(0)
 
     @property
     def shuffle(self):
         """Boolean if shuffle is enabled."""
-        return bool(int(self._status["random"]))
+        return bool(int(self._status.get("random")))
 
     async def async_set_shuffle(self, shuffle: bool) -> None:
         """Enable/disable shuffle mode."""
-        await self._client.random(int(shuffle))
+        async with self.connection():
+            await self._client.random(int(shuffle))
 
     async def async_turn_off(self) -> None:
         """Service to send the MPD the command to stop playing."""
-        await self._client.stop()
+        async with self.connection():
+            await self._client.stop()
 
     async def async_turn_on(self) -> None:
         """Service to send the MPD the command to start playing."""
-        await self._client.play()
-        await self._update_playlists(no_throttle=True)
+        async with self.connection():
+            await self._client.play()
+            await self._update_playlists(no_throttle=True)
 
     async def async_clear_playlist(self) -> None:
         """Clear players playlist."""
-        await self._client.clear()
+        async with self.connection():
+            await self._client.clear()
 
     async def async_media_seek(self, position: float) -> None:
         """Send seek command."""
-        await self._client.seekcur(position)
+        async with self.connection():
+            await self._client.seekcur(position)
 
     async def async_browse_media(
         self,
@@ -513,8 +564,11 @@ class MpdDevice(MediaPlayerEntity):
         media_content_id: str | None = None,
     ) -> BrowseMedia:
         """Implement the websocket media browsing helper."""
-        return await media_source.async_browse_media(
-            self.hass,
-            media_content_id,
-            content_filter=lambda item: item.media_content_type.startswith("audio/"),
-        )
+        async with self.connection():
+            return await media_source.async_browse_media(
+                self.hass,
+                media_content_id,
+                content_filter=lambda item: item.media_content_type.startswith(
+                    "audio/"
+                ),
+            )