diff --git a/neutron/api/rpc/handlers/dhcp_rpc.py b/neutron/api/rpc/handlers/dhcp_rpc.py
index 0b37bbe8e8c..19080534da9 100644
--- a/neutron/api/rpc/handlers/dhcp_rpc.py
+++ b/neutron/api/rpc/handlers/dhcp_rpc.py
@@ -273,6 +273,13 @@ class DhcpRpcCallback(object):
         plugin = directory.get_plugin()
         return self._port_action(plugin, context, port, 'create_port')
 
+    def _is_dhcp_agent_hosting_network(self, plugin, context, host,
+                                       network_id):
+        """Check whether a DHCP agent (host) is hosting a network."""
+        agents = plugin.get_dhcp_agents_hosting_networks(context, [network_id],
+                                                         hosts=[host])
+        return len(agents) != 0
+
     @oslo_messaging.expected_exceptions(exceptions.IpAddressGenerationFailure)
     @db_api.retry_db_errors
     def update_dhcp_port(self, context, **kwargs):
@@ -283,12 +290,14 @@ class DhcpRpcCallback(object):
         port['port'][portbindings.HOST_ID] = host
         plugin = directory.get_plugin()
         try:
+            network_id = port['port']['network_id']
             old_port = plugin.get_port(context, port['id'])
             if (old_port['device_id'] !=
-                constants.DEVICE_ID_RESERVED_DHCP_PORT and
+                    constants.DEVICE_ID_RESERVED_DHCP_PORT and
                 old_port['device_id'] !=
-                utils.get_dhcp_agent_device_id(port['port']['network_id'],
-                                               host)):
+                    utils.get_dhcp_agent_device_id(network_id, host) or
+                not self._is_dhcp_agent_hosting_network(plugin, context, host,
+                    network_id)):
                 raise n_exc.DhcpPortInUse(port_id=port['id'])
             LOG.debug('Update dhcp port %(port)s '
                       'from %(host)s.',
diff --git a/neutron/tests/unit/api/rpc/handlers/test_dhcp_rpc.py b/neutron/tests/unit/api/rpc/handlers/test_dhcp_rpc.py
index 15f9e6abf6a..d3b4ff0a0d8 100644
--- a/neutron/tests/unit/api/rpc/handlers/test_dhcp_rpc.py
+++ b/neutron/tests/unit/api/rpc/handlers/test_dhcp_rpc.py
@@ -43,6 +43,10 @@ class TestDhcpRpcCallback(base.BaseTestCase):
         self.mock_set_dirty = set_dirty_p.start()
         self.utils_p = mock.patch('neutron_lib.plugins.utils.create_port')
         self.utils = self.utils_p.start()
+        self.agent_hosting_network_p = mock.patch.object(self.callbacks,
+            '_is_dhcp_agent_hosting_network')
+        self.mock_agent_hosting_network = self.agent_hosting_network_p.start()
+        self.mock_agent_hosting_network.return_value = True
         self.segment_plugin = mock.MagicMock()
         directory.add_plugin('segments', self.segment_plugin)
 
@@ -328,6 +332,38 @@ class TestDhcpRpcCallback(base.BaseTestCase):
         self.plugin.assert_has_calls([
             mock.call.update_port(mock.ANY, 'foo_port_id', expected_port)])
 
+    def test_update_dhcp_port_with_agent_not_hosting_network(self):
+        port = {'port': {'network_id': 'foo_network_id',
+                         'device_owner': constants.DEVICE_OWNER_DHCP,
+                         'fixed_ips': [{'subnet_id': 'foo_subnet_id'}]}
+                }
+        self.plugin.get_port.return_value = {
+            'device_id': constants.DEVICE_ID_RESERVED_DHCP_PORT}
+        self.mock_agent_hosting_network.return_value = False
+        self.assertRaises(exceptions.DhcpPortInUse,
+                          self.callbacks.update_dhcp_port,
+                          mock.Mock(),
+                          host='foo_host',
+                          port_id='foo_port_id',
+                          port=port)
+
+    def test__is_dhcp_agent_hosting_network(self):
+        self.agent_hosting_network_p.stop()
+        agent = mock.Mock()
+        with mock.patch.object(self.plugin, 'get_dhcp_agents_hosting_networks',
+                               return_value=[agent]):
+            ret = self.callbacks._is_dhcp_agent_hosting_network(self.plugin,
+                mock.Mock(), host='foo_host', network_id='foo_network_id')
+        self.assertTrue(ret)
+
+    def test__is_dhcp_agent_hosting_network_false(self):
+        self.agent_hosting_network_p.stop()
+        with mock.patch.object(self.plugin, 'get_dhcp_agents_hosting_networks',
+                               return_value=[]):
+            ret = self.callbacks._is_dhcp_agent_hosting_network(self.plugin,
+                mock.Mock(), host='foo_host', network_id='foo_network_id')
+        self.assertFalse(ret)
+
     def test_release_dhcp_port(self):
         port_retval = dict(id='port_id', fixed_ips=[dict(subnet_id='a')])
         self.plugin.get_ports.return_value = [port_retval]