diff --git a/mopidy/frontends/mpd/dispatcher.py b/mopidy/frontends/mpd/dispatcher.py
index 6cc05bec..a36d38e3 100644
--- a/mopidy/frontends/mpd/dispatcher.py
+++ b/mopidy/frontends/mpd/dispatcher.py
@@ -20,6 +20,10 @@ from mopidy.utils import flatten
 
 logger = logging.getLogger('mopidy.frontends.mpd.dispatcher')
 
+#: Subsystems that can be registered with idle command.
+SUBSYSTEMS = ['database', 'mixer', 'options', 'output',
+    'player', 'playlist', 'stored_playlist', 'update', ]
+
 class MpdDispatcher(object):
     """
     The MPD session feeds the MPD dispatcher with requests. The dispatcher
@@ -32,6 +36,8 @@ class MpdDispatcher(object):
         self.command_list = False
         self.command_list_ok = False
         self.command_list_index = None
+        self.subscriptions = set()
+        self.events = set()
         self.context = MpdContext(self, session=session)
 
     def handle_request(self, request, current_command_list_index=None):
@@ -42,13 +48,26 @@ class MpdDispatcher(object):
             self._catch_mpd_ack_errors_filter,
             self._authenticate_filter,
             self._command_list_filter,
+            self._idle_filter,
             self._add_ok_filter,
             self._call_handler_filter,
         ]
         return self._call_next_filter(request, response, filter_chain)
 
     def handle_idle(self, subsystem):
-        logger.debug(u'Got idle event for %s', subsystem)
+        self.events.add(subsystem)
+
+        subsystems = self.subscriptions.intersection(self.events)
+        if not subsystems:
+            return
+
+        response = []
+        for subsystem in subsystems:
+            response.append(u'changed: %s' % subsystem)
+        response.append(u'OK')
+        self.subscriptions = set()
+        self.events = set()
+        self.context.session.send_lines(response)
 
     def _call_next_filter(self, request, response, filter_chain):
         if filter_chain:
@@ -111,17 +130,60 @@ class MpdDispatcher(object):
             and request != u'command_list_end')
 
 
+    ### Filter: idle
+
+    def _idle_filter(self, request, response, filter_chain):
+        if re.match(r'^noidle$', request):
+            if not self.subscriptions:
+                return []
+            self.subscriptions = set()
+            self.events = set()
+            self.context.session.connection.enable_timeout()
+            return [u'OK']
+
+        if self.subscriptions:
+            self.context.session.close()
+            return []
+
+        if re.match(r'^idle( .+)?$', request):
+            for subsystem in self._extract_subsystems(request):
+                self.subscriptions.add(subsystem)
+
+            subsystems = self.subscriptions.intersection(self.events)
+            if subsystems:
+                for subsystem in subsystems:
+                    response.append(u'changed: %s' % subsystem)
+                self.events = set()
+                self.subscriptions = set()
+                response.append(u'OK')
+                return response
+            else:
+                self.context.session.connection.disable_timeout()
+                return []
+
+        return self._call_next_filter(request, response, filter_chain)
+
+    def _extract_subsystems(self, request):
+        match = re.match(r'^idle (?P<subsystems>.+)$', request)
+        if not match:
+            return SUBSYSTEMS
+        return match.groupdict()['subsystems'].split(' ')
+
+
     ### Filter: add OK
 
     def _add_ok_filter(self, request, response, filter_chain):
         response = self._call_next_filter(request, response, filter_chain)
-        if not self._has_error(response):
+        if not self._has_error(response) and not self._is_idle(request):
             response.append(u'OK')
         return response
 
     def _has_error(self, response):
         return response and response[-1].startswith(u'ACK')
 
+    def _is_idle(self, request):
+        return request.startswith('idle')
+
 
     ### Filter: call handler
 
diff --git a/mopidy/frontends/mpd/protocol/status.py b/mopidy/frontends/mpd/protocol/status.py
index abbb8d7f..4a961e76 100644
--- a/mopidy/frontends/mpd/protocol/status.py
+++ b/mopidy/frontends/mpd/protocol/status.py
@@ -32,8 +32,8 @@ def currentsong(context):
             position=context.backend.playback.current_playlist_position.get(),
             cpid=current_cp_track.cpid)
 
-@handle_request(r'^idle$')
-@handle_request(r'^idle (?P<subsystems>.+)$')
+#@handle_request(r'^idle$')
+#@handle_request(r'^idle (?P<subsystems>.+)$')
 def idle(context, subsystems=None):
     """
     *musicpd.org, status section:*
@@ -69,7 +69,7 @@ def idle(context, subsystems=None):
     """
     pass # TODO
 
-@handle_request(r'^noidle$')
+#@handle_request(r'^noidle$')
 def noidle(context):
     """See :meth:`_status_idle`."""
     pass # TODO
diff --git a/tests/frontends/mpd/protocol/__init__.py b/tests/frontends/mpd/protocol/__init__.py
index 77825a6e..68668bc2 100644
--- a/tests/frontends/mpd/protocol/__init__.py
+++ b/tests/frontends/mpd/protocol/__init__.py
@@ -27,6 +27,7 @@ class BaseTestCase(unittest.TestCase):
         self.connection = MockConnetion()
         self.session = mpd.MpdSession(self.connection)
         self.dispatcher = self.session.dispatcher
+        self.context = self.dispatcher.context
 
     def tearDown(self):
         self.backend.stop().get()
@@ -38,6 +39,9 @@ class BaseTestCase(unittest.TestCase):
         self.session.on_line_received(request)
         return self.connection.response
 
+    def assertNoResponse(self):
+        self.assertEqual([], self.connection.response)
+
     def assertInResponse(self, value):
         self.assert_(value in self.connection.response, u'Did not find %s '
             'in %s' % (repr(value), repr(self.connection.response)))
diff --git a/tests/frontends/mpd/protocol/idle_test.py b/tests/frontends/mpd/protocol/idle_test.py
new file mode 100644
index 00000000..319789bf
--- /dev/null
+++ b/tests/frontends/mpd/protocol/idle_test.py
@@ -0,0 +1,210 @@
+from mock import patch
+
+from mopidy.frontends.mpd.dispatcher import SUBSYSTEMS
+from mopidy.models import Track
+
+from tests.frontends.mpd import protocol
+
+class IdleHandlerTest(protocol.BaseTestCase):
+    def idleEvent(self, subsystem):
+        self.session.on_idle(subsystem)
+
+    def assertEqualEvents(self, events):
+        self.assertEqual(set(events), self.dispatcher.events)
+
+    def assertEqualSubscriptions(self, events):
+        self.assertEqual(set(events), self.dispatcher.subscriptions)
+
+    def assertNoEvents(self):
+        self.assertEqualEvents([])
+
+    def assertNoSubscriptions(self):
+        self.assertEqualSubscriptions([])
+
+    def test_base_state(self):
+        self.assertNoSubscriptions()
+        self.assertNoEvents()
+        self.assertNoResponse()
+
+    def test_idle(self):
+        self.sendRequest(u'idle')
+        self.assertEqualSubscriptions(SUBSYSTEMS)
+        self.assertNoEvents()
+        self.assertNoResponse()
+
+    def test_idle_disables_timeout(self):
+        self.sendRequest(u'idle')
+        self.connection.disable_timeout.assert_called_once_with()
+
+    def test_noidle(self):
+        self.sendRequest(u'noidle')
+        self.assertNoSubscriptions()
+        self.assertNoEvents()
+        self.assertNoResponse()
+
+    def test_noidle_does_not_call_enable_timeout(self):
+        self.sendRequest(u'noidle')
+        self.assertEqual(0, self.connection.enable_timeout.call_count)
+
+    def test_idle_player(self):
+        self.sendRequest(u'idle player')
+        self.assertEqualSubscriptions(['player'])
+        self.assertNoEvents()
+        self.assertNoResponse()
+
+    def test_idle_player_playlist(self):
+        self.sendRequest(u'idle player playlist')
+        self.assertEqualSubscriptions(['player', 'playlist'])
+        self.assertNoEvents()
+        self.assertNoResponse()
+
+    def test_idle_then_noidle(self):
+        self.sendRequest(u'idle')
+        self.sendRequest(u'noidle')
+        self.assertNoSubscriptions()
+        self.assertNoEvents()
+        self.assertInResponse(u'OK')
+
+    def test_idle_then_noidle_enables_timeout(self):
+        self.sendRequest(u'idle')
+        self.sendRequest(u'noidle')
+        self.connection.enable_timeout.assert_called_once_with()
+
+    def test_idle_then_play(self):
+        with patch.object(self.session, 'stop') as stop_mock:
+            self.sendRequest(u'idle')
+            self.sendRequest(u'play')
+            stop_mock.assert_called_once_with()
+
+    def test_idle_then_idle(self):
+        with patch.object(self.session, 'stop') as stop_mock:
+            self.sendRequest(u'idle')
+            self.sendRequest(u'idle')
+            stop_mock.assert_called_once_with()
+
+    def test_idle_player_then_play(self):
+        with patch.object(self.session, 'stop') as stop_mock:
+            self.sendRequest(u'idle player')
+            self.sendRequest(u'play')
+            stop_mock.assert_called_once_with()
+
+    def test_idle_then_player(self):
+        self.sendRequest(u'idle')
+        self.idleEvent(u'player')
+        self.assertNoSubscriptions()
+        self.assertNoEvents()
+        self.assertInResponse(u'changed: player')
+        self.assertInResponse(u'OK')
+
+    def test_idle_player_then_event_player(self):
+        self.sendRequest(u'idle player')
+        self.idleEvent(u'player')
+        self.assertNoSubscriptions()
+        self.assertNoEvents()
+        self.assertInResponse(u'changed: player')
+        self.assertInResponse(u'OK')
+
+    def test_idle_player_then_noidle(self):
+        self.sendRequest(u'idle player')
+        self.sendRequest(u'noidle')
+        self.assertNoSubscriptions()
+        self.assertNoEvents()
+        self.assertInResponse(u'OK')
+
+    def test_idle_player_playlist_then_noidle(self):
+        self.sendRequest(u'idle player playlist')
+        self.sendRequest(u'noidle')
+        self.assertNoEvents()
+        self.assertNoSubscriptions()
+        self.assertInResponse(u'OK')
+
+    def test_idle_player_playlist_then_player(self):
+        self.sendRequest(u'idle player playlist')
+        self.idleEvent(u'player')
+        self.assertNoEvents()
+        self.assertNoSubscriptions()
+        self.assertInResponse(u'changed: player')
+        self.assertNotInResponse(u'changed: playlist')
+        self.assertInResponse(u'OK')
+
+    def test_idle_playlist_then_player(self):
+        self.sendRequest(u'idle playlist')
+        self.idleEvent(u'player')
+        self.assertEqualEvents(['player'])
+        self.assertEqualSubscriptions(['playlist'])
+        self.assertNoResponse()
+
+    def test_idle_playlist_then_player_then_playlist(self):
+        self.sendRequest(u'idle playlist')
+        self.idleEvent(u'player')
+        self.idleEvent(u'playlist')
+        self.assertNoEvents()
+        self.assertNoSubscriptions()
+        self.assertNotInResponse(u'changed: player')
+        self.assertInResponse(u'changed: playlist')
+        self.assertInResponse(u'OK')
+
+    def test_player(self):
+        self.idleEvent(u'player')
+        self.assertEqualEvents(['player'])
+        self.assertNoSubscriptions()
+        self.assertNoResponse()
+
+    def test_player_then_idle_player(self):
+        self.idleEvent(u'player')
+        self.sendRequest(u'idle player')
+        self.assertNoEvents()
+        self.assertNoSubscriptions()
+        self.assertInResponse(u'changed: player')
+        self.assertNotInResponse(u'changed: playlist')
+        self.assertInResponse(u'OK')
+
+    def test_player_then_playlist(self):
+        self.idleEvent(u'player')
+        self.idleEvent(u'playlist')
+        self.assertEqualEvents(['player', 'playlist'])
+        self.assertNoSubscriptions()
+        self.assertNoResponse()
+
+    def test_player_then_idle(self):
+        self.idleEvent(u'player')
+        self.sendRequest(u'idle')
+        self.assertNoEvents()
+        self.assertNoSubscriptions()
+        self.assertInResponse(u'changed: player')
+        self.assertInResponse(u'OK')
+
+    def test_player_then_playlist_then_idle(self):
+        self.idleEvent(u'player')
+        self.idleEvent(u'playlist')
+        self.sendRequest(u'idle')
+        self.assertNoEvents()
+        self.assertNoSubscriptions()
+        self.assertInResponse(u'changed: player')
+        self.assertInResponse(u'changed: playlist')
+        self.assertInResponse(u'OK')
+
+    def test_player_then_idle_playlist(self):
+        self.idleEvent(u'player')
+        self.sendRequest(u'idle playlist')
+        self.assertEqualEvents(['player'])
+        self.assertEqualSubscriptions(['playlist'])
+        self.assertNoResponse()
+
+    def test_player_then_idle_playlist_then_noidle(self):
+        self.idleEvent(u'player')
+        self.sendRequest(u'idle playlist')
+        self.sendRequest(u'noidle')
+        self.assertNoEvents()
+        self.assertNoSubscriptions()
+        self.assertInResponse(u'OK')
+
+    def test_player_then_playlist_then_idle_playlist(self):
+        self.idleEvent(u'player')
+        self.idleEvent(u'playlist')
+        self.sendRequest(u'idle playlist')
+        self.assertNoEvents()
+        self.assertNoSubscriptions()
+        self.assertNotInResponse(u'changed: player')
+        self.assertInResponse(u'changed: playlist')
+        self.assertInResponse(u'OK')
diff --git a/tests/frontends/mpd/protocol/reflection_test.py b/tests/frontends/mpd/protocol/reflection_test.py
index 315e3051..9cf1ef97 100644
--- a/tests/frontends/mpd/protocol/reflection_test.py
+++ b/tests/frontends/mpd/protocol/reflection_test.py
@@ -9,14 +9,14 @@ class ReflectionHandlerTest(protocol.BaseTestCase):
         self.assertInResponse(u'command: commands')
         self.assertInResponse(u'command: play')
         self.assertInResponse(u'command: status')
+        self.assertInResponse(u'command: idle')
+        self.assertInResponse(u'command: noidle')
         # Check if commands you do not have access to are not present
         self.assertNotInResponse(u'command: kill')
         # Check if the blacklisted commands are not present
         self.assertNotInResponse(u'command: command_list_begin')
         self.assertNotInResponse(u'command: command_list_ok_begin')
         self.assertNotInResponse(u'command: command_list_end')
-        self.assertNotInResponse(u'command: idle')
-        self.assertNotInResponse(u'command: noidle')
         self.assertNotInResponse(u'command: sticker')
         self.assertInResponse(u'OK')
 
diff --git a/tests/frontends/mpd/protocol/status_test.py b/tests/frontends/mpd/protocol/status_test.py
index 6762a4fb..f50ecd24 100644
--- a/tests/frontends/mpd/protocol/status_test.py
+++ b/tests/frontends/mpd/protocol/status_test.py
@@ -27,21 +27,6 @@ class StatusHandlerTest(protocol.BaseTestCase):
         self.sendRequest(u'currentsong')
         self.assertInResponse(u'OK')
 
-    def test_idle_without_subsystems(self):
-        # FIXME this is not the correct behaviour for idle...
-        self.sendRequest(u'idle')
-        self.assertInResponse(u'OK')
-
-    def test_idle_with_subsystems(self):
-        # FIXME this is not the correct behaviour for idle...
-        self.sendRequest(u'idle database playlist')
-        self.assertInResponse(u'OK')
-
-    def test_noidle(self):
-        # FIXME this is not the correct behaviour for idle...
-        self.sendRequest(u'noidle')
-        self.assertInResponse(u'OK')
-
     def test_stats_command(self):
         self.sendRequest(u'stats')
         self.assertInResponse(u'OK')