diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 7c4898c6f..d4e6425e8 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -13,7 +13,7 @@ jobs:
       fail-fast: false
       matrix:
         os: [ubuntu, macos, windows]
-        python-version: [ '3.6', '3.7', '3.8', '3.9', 'pypy3' ]
+        python-version: [ '3.7', '3.8', '3.9' ]
         exclude:
         - os: windows
           python-version: pypy3
diff --git a/ipykernel/compiler.py b/ipykernel/compiler.py
new file mode 100644
index 000000000..4c724f146
--- /dev/null
+++ b/ipykernel/compiler.py
@@ -0,0 +1,64 @@
+from IPython.core.compilerop import CachingCompiler
+import tempfile
+import os
+
+def murmur2_x86(data, seed):
+    m = 0x5bd1e995
+    length = len(data)
+    h = seed ^ length
+    rounded_end = (length & 0xfffffffc)
+    for i in range(0, rounded_end, 4):
+        k = (ord(data[i]) & 0xff) | ((ord(data[i + 1]) & 0xff) << 8) | \
+           ((ord(data[i + 2]) & 0xff) << 16) | (ord(data[i + 3]) << 24)
+        k = (k * m) & 0xffffffff
+        k ^= k >> 24
+        k = (k * m) & 0xffffffff
+
+        h = (h * m) & 0xffffffff
+        h ^= k
+
+    val = length & 0x03
+    k = 0
+    if val == 3:
+        k = (ord(data[rounded_end + 2]) & 0xff) << 16
+    if val in [2, 3]:
+        k |= (ord(data[rounded_end + 1]) & 0xff) << 8
+    if val in [1, 2, 3]:
+        k |= ord(data[rounded_end]) & 0xff
+        h ^= k
+        h = (h * m) & 0xffffffff
+
+    h ^= h >> 13
+    h = (h * m) & 0xffffffff
+    h ^= h >> 15
+
+    return h
+
+def get_tmp_directory():
+    tmp_dir = tempfile.gettempdir()
+    pid = os.getpid()
+    return tmp_dir + '/ipykernel_' + str(pid)
+
+def get_tmp_hash_seed():
+    hash_seed = 0xc70f6907
+    return hash_seed
+
+def get_file_name(code):
+    name = murmur2_x86(code, get_tmp_hash_seed())
+    return get_tmp_directory() + '/' + str(name) + '.py'
+
+class XCachingCompiler(CachingCompiler):
+
+    def __init__(self, *args, **kwargs):
+        super(XCachingCompiler, self).__init__(*args, **kwargs)
+        self.filename_mapper = None
+        self.log = None
+
+    def get_code_name(self, raw_code, code, number):
+        filename = get_file_name(raw_code)
+
+        if self.filename_mapper is not None:
+            self.filename_mapper(filename, number)
+
+        return filename
+
diff --git a/ipykernel/control.py b/ipykernel/control.py
index 10853046d..7ac56b7f9 100644
--- a/ipykernel/control.py
+++ b/ipykernel/control.py
@@ -12,6 +12,8 @@ class ControlThread(Thread):
     def __init__(self, **kwargs):
         Thread.__init__(self, **kwargs)
         self.io_loop = IOLoop(make_current=False)
+        self.pydev_do_not_trace = True
+        self.is_pydev_daemon_thread = True
 
     def run(self): 
         self.io_loop.make_current()
diff --git a/ipykernel/debugger.py b/ipykernel/debugger.py
new file mode 100644
index 000000000..e2bc8fc49
--- /dev/null
+++ b/ipykernel/debugger.py
@@ -0,0 +1,426 @@
+import logging
+import os
+
+import zmq
+from zmq.utils import jsonapi
+
+from tornado.queues import Queue
+from tornado.locks import Event
+
+from .compiler import (get_file_name, get_tmp_directory, get_tmp_hash_seed)
+
+import debugpy
+
+class DebugpyMessageQueue:
+
+    HEADER = 'Content-Length: '
+    HEADER_LENGTH = 16
+    SEPARATOR = '\r\n\r\n'
+    SEPARATOR_LENGTH = 4
+
+    def __init__(self, event_callback, log):
+        self.tcp_buffer = ''
+        self._reset_tcp_pos()
+        self.event_callback = event_callback
+        self.message_queue = Queue()
+        self.log = log
+
+    def _reset_tcp_pos(self):
+        self.header_pos = -1
+        self.separator_pos = -1
+        self.message_size = 0
+        self.message_pos = -1
+
+    def _put_message(self, raw_msg):
+        self.log.debug('QUEUE - _put_message:')
+        msg = jsonapi.loads(raw_msg)
+        if msg['type'] == 'event':
+            self.log.debug('QUEUE - received event:')
+            self.log.debug(msg)
+            self.event_callback(msg)
+        else:
+            self.log.debug('QUEUE - put message:')
+            self.log.debug(msg)
+            self.message_queue.put_nowait(msg)
+
+    def put_tcp_frame(self, frame):
+        self.tcp_buffer += frame
+
+        self.log.debug('QUEUE - received frame')
+        while True:
+            # Finds header
+            if self.header_pos == -1:
+                self.header_pos = self.tcp_buffer.find(DebugpyMessageQueue.HEADER)
+            if self.header_pos == -1:
+                return
+        
+            self.log.debug('QUEUE - found header at pos %i', self.header_pos)
+
+            #Finds separator
+            if self.separator_pos == -1:
+                hint = self.header_pos + DebugpyMessageQueue.HEADER_LENGTH
+                self.separator_pos = self.tcp_buffer.find(DebugpyMessageQueue.SEPARATOR, hint)
+            if self.separator_pos == -1:
+                return
+
+            self.log.debug('QUEUE - found separator at pos %i', self.separator_pos)
+
+            if self.message_pos == -1:
+                size_pos = self.header_pos + DebugpyMessageQueue.HEADER_LENGTH
+                self.message_pos = self.separator_pos + DebugpyMessageQueue.SEPARATOR_LENGTH
+                self.message_size = int(self.tcp_buffer[size_pos:self.separator_pos])
+
+            self.log.debug('QUEUE - found message at pos %i', self.message_pos)
+            self.log.debug('QUEUE - message size is %i', self.message_size)
+
+            if len(self.tcp_buffer) - self.message_pos < self.message_size:
+                return
+
+            self._put_message(self.tcp_buffer[self.message_pos:self.message_pos + self.message_size])
+            if len(self.tcp_buffer) - self.message_pos == self.message_size:
+                self.log.debug('QUEUE - resetting tcp_buffer')
+                self.tcp_buffer = ''
+                self._reset_tcp_pos()
+                return
+            else:
+                self.tcp_buffer = self.tcp_buffer[self.message_pos + self.message_size:]
+                self.log.debug('QUEUE - slicing tcp_buffer: %s', self.tcp_buffer)
+                self._reset_tcp_pos()
+
+    async def get_message(self):
+        return await self.message_queue.get()
+            
+
+class DebugpyClient:
+
+    def __init__(self, log, debugpy_stream, event_callback):
+        self.log = log
+        self.debugpy_stream = debugpy_stream
+        self.event_callback = event_callback
+        self.message_queue = DebugpyMessageQueue(self._forward_event, self.log)
+        self.debugpy_host = '127.0.0.1'
+        self.debugpy_port = -1
+        self.routing_id = None
+        self.wait_for_attach = True
+        self.init_event = Event()
+        self.init_event_seq = -1
+
+    def _get_endpoint(self):
+        host, port = self.get_host_port()
+        return 'tcp://' + host + ':' + str(port)
+
+    def _forward_event(self, msg):
+        if msg['event'] == 'initialized':
+            self.init_event.set()
+            self.init_event_seq = msg['seq']
+        self.event_callback(msg)
+
+    def _send_request(self, msg):
+        if self.routing_id is None:
+            self.routing_id = self.debugpy_stream.socket.getsockopt(zmq.ROUTING_ID)
+        content = jsonapi.dumps(msg)
+        content_length = str(len(content))
+        buf = (DebugpyMessageQueue.HEADER + content_length + DebugpyMessageQueue.SEPARATOR).encode('ascii')
+        buf += content
+        self.log.debug("DEBUGPYCLIENT:")
+        self.log.debug(self.routing_id)
+        self.log.debug(buf)
+        self.debugpy_stream.send_multipart((self.routing_id, buf))
+    
+    async def _wait_for_response(self):
+        # Since events are never pushed to the message_queue
+        # we can safely assume the next message in queue
+        # will be an answer to the previous request
+        return await self.message_queue.get_message()
+
+    async def _handle_init_sequence(self):
+        # 1] Waits for initialized event
+        await self.init_event.wait()
+        
+        # 2] Sends configurationDone request
+        configurationDone = {
+            'type': 'request',
+            'seq': int(self.init_event_seq) + 1,
+            'command': 'configurationDone'
+        }
+        self._send_request(configurationDone)
+
+        # 3]  Waits for configurationDone response
+        await self._wait_for_response()
+
+        # 4] Waits for attachResponse and returns it
+        attach_rep = await self._wait_for_response()
+        return attach_rep
+
+    def get_host_port(self):
+        if self.debugpy_port == -1:
+            socket = self.debugpy_stream.socket
+            socket.bind_to_random_port('tcp://' + self.debugpy_host)
+            self.endpoint = socket.getsockopt(zmq.LAST_ENDPOINT).decode('utf-8')
+            socket.unbind(self.endpoint)
+            index = self.endpoint.rfind(':')
+            self.debugpy_port = self.endpoint[index+1:]
+        return self.debugpy_host, self.debugpy_port
+
+    def connect_tcp_socket(self):
+        self.debugpy_stream.socket.connect(self._get_endpoint())
+        self.routing_id = self.debugpy_stream.socket.getsockopt(zmq.ROUTING_ID)
+
+    def disconnect_tcp_socket(self):
+        self.debugpy_stream.socket.disconnect(self._get_endpoint())
+        self.routing_id = None
+        self.init_event = Event()
+        self.init_event_seq = -1
+        self.wait_for_attach = True
+
+    def receive_dap_frame(self, frame):
+        self.message_queue.put_tcp_frame(frame)
+
+    async def send_dap_request(self, msg):
+        self._send_request(msg)
+        if self.wait_for_attach and msg['command'] == 'attach':
+            rep = await self._handle_init_sequence()
+            self.wait_for_attach = False
+            return rep
+        else:
+            rep = await self._wait_for_response()
+            self.log.debug('DEBUGPYCLIENT - returning:')
+            self.log.debug(rep)
+            return rep
+
+class Debugger:
+
+    # Requests that requires that the debugger has started
+    started_debug_msg_types = [
+        'dumpCell', 'setBreakpoints',
+        'source', 'stackTrace',
+        'variables', 'attach',
+        'configurationDone'
+    ]
+    
+    # Requests that can be handled even if the debugger is not running
+    static_debug_msg_types = [
+        'debugInfo', 'inspectVariables'
+    ]
+
+    def __init__(self, log, debugpy_stream, event_callback, shell_socket, session):
+        self.log = log
+        self.debugpy_client = DebugpyClient(log, debugpy_stream, self._handle_event)
+        self.shell_socket = shell_socket
+        self.session = session
+        self.is_started = False
+        self.event_callback = event_callback
+        
+        self.started_debug_handlers = {}
+        for msg_type in Debugger.started_debug_msg_types:
+            self.started_debug_handlers[msg_type] = getattr(self, msg_type)
+
+        self.static_debug_handlers = {}
+        for msg_type in Debugger.static_debug_msg_types:
+            self.static_debug_handlers[msg_type] = getattr(self, msg_type)
+
+        self.breakpoint_list = {}
+        self.stopped_threads = []
+
+        self.debugpy_initialized = False
+
+        self.debugpy_host = '127.0.0.1'
+        self.debugpy_port = 0
+        self.endpoint = None
+
+    def _handle_event(self, msg):
+        if msg['event'] == 'stopped':
+            self.stopped_threads.append(msg['body']['threadId'])
+        elif msg['event'] == 'continued':
+            try:
+                self.stopped_threads.remove(msg['body']['threadId'])
+            except:
+                pass
+        self.event_callback(msg)
+
+    async def _forward_message(self, msg):
+        return await self.debugpy_client.send_dap_request(msg)
+
+    @property
+    def tcp_client(self):
+        return self.debugpy_client
+
+    def start(self):
+        if not self.debugpy_initialized:
+            tmp_dir = get_tmp_directory()
+            if not os.path.exists(tmp_dir):
+                os.makedirs(tmp_dir)
+            host, port = self.debugpy_client.get_host_port()
+            code = 'import debugpy;'
+            code += 'debugpy.listen(("' + host + '",' + port + '))'
+            content = {
+                'code': code,
+                'silent': True
+            }
+            self.session.send(self.shell_socket, 'execute_request', content,
+                              None, (self.shell_socket.getsockopt(zmq.ROUTING_ID)))
+                          
+            ident, msg = self.session.recv(self.shell_socket, mode=0)
+            self.debugpy_initialized = msg['content']['status'] == 'ok'
+        self.debugpy_client.connect_tcp_socket()
+        return self.debugpy_initialized
+
+    def stop(self):
+        self.debugpy_client.disconnect_tcp_socket()
+
+    async def dumpCell(self, message):
+        code = message['arguments']['code']
+        file_name = get_file_name(code)
+
+        with open(file_name, 'w') as f:
+            f.write(code)
+
+        reply = {
+            'type': 'response',
+            'request_seq': message['seq'],
+            'success': True,
+            'command': message['command'],
+            'body': {
+                'sourcePath': file_name
+            }
+        }
+        return reply
+
+    async def setBreakpoints(self, message):
+        source = message['arguments']['source']['path'];
+        self.breakpoint_list[source] = message['arguments']['breakpoints']
+        return await self._forward_message(message);
+
+    async def source(self, message):
+        reply = {
+            'type': 'response',
+            'request_seq': message['seq'],
+            'command': message['command']
+        }
+        source_path = message['arguments']['source']['path'];
+        if os.path.isfile(source_path):
+            with open(source_path) as f:
+                reply['success'] = True
+                reply['body'] = {
+                    'content': f.read()
+                }
+        else:
+            reply['success'] = False
+            reply['message'] = 'source unavailable'
+            reply['body'] = {}
+
+        return reply
+
+    async def stackTrace(self, message):
+        reply = await self._forward_message(message)
+        # The stackFrames array has the following content:
+        # { frames from the notebook}
+        # ...
+        # { 'id': xxx, 'name': '<module>', ... } <= this is the first frame of the code from the notebook
+        # { frames from ipykernel }
+        # ...
+        # {'id': yyy, 'name': '<module>', ... } <= this is the first frame of ipykernel code
+        # We want to remove all the frames from ipykernel
+        sf_list = reply['body']['stackFrames']
+        module_idx = len(sf_list) - next(i for i, v in enumerate(reversed(sf_list), 1) if v['name'] == '<module>' and i != 1)
+        reply['body']['stackFrames'] = reply['body']['stackFrames'][:module_idx+1]
+        return reply
+
+    def accept_variable(self, variable):
+        cond = variable['type'] != 'list' and variable['type'] != 'ZMQExitAutocall' and variable['type'] != 'dict'
+        cond = cond and variable['name'] not in ['debugpy', 'get_ipython', '_']
+        cond = cond and variable['name'][0:2] != '_i'
+        return cond
+
+    async def variables(self, message):
+        reply = await self._forward_message(message)
+        # TODO : check start and count arguments work as expected in debugpy
+        reply['body']['variables'] = \
+            [var for var in reply['body']['variables'] if self.accept_variable(var)]
+        return reply
+
+    async def attach(self, message):
+        host, port = self.debugpy_client.get_host_port()
+        message['arguments']['connect'] = {
+            'host': host,
+            'port': port
+        }
+        message['arguments']['logToFile'] = True
+        return await self._forward_message(message)
+
+    async def configurationDone(self, message):
+        reply = {
+            'seq': message['seq'],
+            'type': 'response',
+            'request_seq': message['seq'],
+            'success': True,
+            'command': message['command']
+        }
+        return reply;
+
+    async def debugInfo(self, message):
+        breakpoint_list = []
+        for key, value in self.breakpoint_list.items():
+            breakpoint_list.append({
+                'source': key,
+                'breakpoints': value
+            })
+        reply = {
+            'type': 'response',
+            'request_seq': message['seq'],
+            'success': True,
+            'command': message['command'],
+            'body': {
+                'isStarted': self.is_started,
+                'hashMethod': 'Murmur2',
+                'hashSeed': get_tmp_hash_seed(),
+                'tmpFilePrefix': get_tmp_directory() + '/',
+                'tmpFileSuffix': '.py',
+                'breakpoints': breakpoint_list,
+                'stoppedThreads': self.stopped_threads
+            }
+        }
+        return reply
+
+    async def inspectVariables(self, message):
+        # TODO
+        return {}
+
+    async def process_request(self, message):
+        reply = {}
+
+        if message['command'] == 'initialize':
+            if self.is_started:
+                self.log.info('The debugger has already started')
+            else:
+                self.is_started = self.start()
+                if self.is_started:
+                    self.log.info('The debugger has started')
+                else:
+                    reply = {
+                        'command', 'initialize',
+                        'request_seq', message['seq'],
+                        'seq', 3,
+                        'success', False,
+                        'type', 'response'
+                    }
+
+        handler = self.static_debug_handlers.get(message['command'], None)
+        if handler is not None:
+            reply = await handler(message)
+        elif self.is_started:
+            handler = self.started_debug_handlers.get(message['command'], None)
+            if handler is not None:
+                reply = await handler(message)
+            else:
+                reply = await self._forward_message(message)
+
+        if message['command'] == 'disconnect':
+            self.stop()
+            self.breakpoint_list = {}
+            self.stopped_threads = []
+            self.is_started = False
+            self.log.info('The debugger has stopped')
+
+        return reply
+
diff --git a/ipykernel/heartbeat.py b/ipykernel/heartbeat.py
index 01be21122..6d65a6c7e 100644
--- a/ipykernel/heartbeat.py
+++ b/ipykernel/heartbeat.py
@@ -40,6 +40,8 @@ def __init__(self, context, addr=None):
             self.pick_port()
         self.addr = (self.ip, self.port)
         self.daemon = True
+        self.pydev_do_not_trace = True
+        self.is_pydev_daemon_thread = True
 
     def pick_port(self):
         if self.transport == 'tcp':
diff --git a/ipykernel/iostream.py b/ipykernel/iostream.py
index 486e80e27..cb3382604 100644
--- a/ipykernel/iostream.py
+++ b/ipykernel/iostream.py
@@ -71,6 +71,8 @@ def __init__(self, socket, pipe=False):
         self._setup_event_pipe()
         self.thread = threading.Thread(target=self._thread_main)
         self.thread.daemon = True
+        self.thread.pydev_do_not_trace = True
+        self.thread.is_pydev_daemon_thread = True
 
     def _thread_main(self):
         """The inner loop that's actually run in a thread"""
diff --git a/ipykernel/ipkernel.py b/ipykernel/ipkernel.py
index 62628971a..23c06f034 100644
--- a/ipykernel/ipkernel.py
+++ b/ipykernel/ipkernel.py
@@ -19,6 +19,8 @@
 from .zmqshell import ZMQInteractiveShell
 from .eventloops import _use_appnope
 
+from .compiler import XCachingCompiler
+
 try:
     from IPython.core.interactiveshell import _asyncio_runner
 except ImportError:
@@ -71,6 +73,7 @@ def __init__(self, **kwargs):
             user_module = self.user_module,
             user_ns     = self.user_ns,
             kernel      = self,
+            compiler_class = XCachingCompiler,
         )
         self.shell.displayhook.session = self.session
         self.shell.displayhook.pub_socket = self.iopub_socket
@@ -150,7 +153,8 @@ def set_parent(self, ident, parent, channel='shell'):
         about the parent message.
         """
         super(IPythonKernel, self).set_parent(ident, parent, channel)
-        self.shell.set_parent(parent)
+        if channel == 'shell':
+            self.shell.set_parent(parent)
 
     def init_metadata(self, parent):
         """Initialize metadata.
diff --git a/ipykernel/kernelapp.py b/ipykernel/kernelapp.py
index c7756fa2c..7bf4a54ba 100644
--- a/ipykernel/kernelapp.py
+++ b/ipykernel/kernelapp.py
@@ -122,6 +122,8 @@ class IPKernelApp(BaseIPythonApplication, InteractiveShellApp,
     context = Any()
     shell_socket = Any()
     control_socket = Any()
+    debugpy_socket = Any()
+    debug_shell_socket = Any()
     stdin_socket = Any()
     iopub_socket = Any()
     iopub_thread = Any()
@@ -294,6 +296,14 @@ def init_control(self, context):
         self.control_port = self._bind_socket(self.control_socket, self.control_port)
         self.log.debug("control ROUTER Channel on port: %i" % self.control_port)
 
+        self.debugpy_socket = context.socket(zmq.STREAM)
+        self.debugpy_socket.linger = 1000
+
+        self.debug_shell_socket = context.socket(zmq.DEALER)
+        self.debug_shell_socket.linger = 1000
+        if self.shell_socket.getsockopt(zmq.LAST_ENDPOINT):
+            self.debug_shell_socket.connect(self.shell_socket.getsockopt(zmq.LAST_ENDPOINT))
+
         if hasattr(zmq, 'ROUTER_HANDOVER'):
             # set router-handover to workaround zeromq reconnect problems
             # in certain rare circumstances
@@ -335,6 +345,12 @@ def close(self):
             self.log.debug("Closing iopub channel")
             self.iopub_thread.stop()
             self.iopub_thread.close()
+
+        if self.debugpy_socket and not self.debugpy_socket.closed:
+            self.debugpy_socket.close()
+        if self.debug_shell_socket and not self.debug_shell_socket.closed:
+            self.debug_shell_socket.close()
+
         for channel in ('shell', 'control', 'stdin'):
             self.log.debug("Closing %s channel", channel)
             socket = getattr(self, channel + "_socket", None)
@@ -449,12 +465,16 @@ def init_kernel(self):
         """Create the Kernel object itself"""
         shell_stream = ZMQStream(self.shell_socket)
         control_stream = ZMQStream(self.control_socket, self.control_thread.io_loop)
+        debugpy_stream = ZMQStream(self.debugpy_socket, self.control_thread.io_loop)
         self.control_thread.start()
         kernel_factory = self.kernel_class.instance
 
         kernel = kernel_factory(parent=self, session=self.session,
                                 control_stream=control_stream,
+                                debugpy_stream=debugpy_stream,
+                                debug_shell_socket=self.debug_shell_socket,
                                 shell_stream=shell_stream, 
+                                control_thread=self.control_thread,
                                 iopub_thread=self.iopub_thread,
                                 iopub_socket=self.iopub_socket,
                                 stdin_socket=self.stdin_socket,
diff --git a/ipykernel/kernelbase.py b/ipykernel/kernelbase.py
index 4d26a73a7..7b2917d23 100644
--- a/ipykernel/kernelbase.py
+++ b/ipykernel/kernelbase.py
@@ -12,6 +12,7 @@
 import time
 import uuid
 import warnings
+import asyncio
 
 try:
     # jupyter_client >= 5, use tz-aware now
@@ -39,6 +40,7 @@
 
 from ._version import kernel_protocol_version
 
+from .debugger import Debugger
 
 class Kernel(SingletonConfigurable):
 
@@ -69,6 +71,11 @@ def shell_streams(self):
         return [shell_stream]
 
     control_stream = Instance(ZMQStream, allow_none=True)
+    debugpy_stream = Instance(ZMQStream, allow_none=True)
+
+    debug_shell_socket = Any()
+
+    control_thread = Any()
     iopub_socket = Any()
     iopub_thread = Any()
     stdin_socket = Any()
@@ -160,7 +167,7 @@ def _default_ident(self):
         'apply_request',
     ]
     # add deprecated ipyparallel control messages
-    control_msg_types = msg_types + ['clear_request', 'abort_request']
+    control_msg_types = msg_types + ['clear_request', 'abort_request', 'debug_request']
 
     def __init__(self, **kwargs):
         super(Kernel, self).__init__(**kwargs)
@@ -173,8 +180,35 @@ def __init__(self, **kwargs):
         for msg_type in self.control_msg_types:
             self.control_handlers[msg_type] = getattr(self, msg_type)
 
+        self.debugger = Debugger(self.log,
+                                 self.debugpy_stream,
+                                 self._publish_debug_event,
+                                 self.debug_shell_socket,
+                                 self.session)
+
+        self.control_queue = Queue()
+        if 'control_thread' in kwargs:
+            kwargs['control_thread'].io_loop.add_callback(self.poll_control_queue)
+
+    @gen.coroutine
+    def dispatch_debugpy(self, msg):
+        # The first frame is the socket id, we can drop it
+        frame = msg[1].bytes.decode('utf-8')
+        self.log.debug("Debugpy received: %s", frame)
+        self.debugger.tcp_client.receive_dap_frame(frame)
+
     @gen.coroutine
     def dispatch_control(self, msg):
+        self.control_queue.put_nowait(msg)
+
+    @gen.coroutine
+    def poll_control_queue(self):
+        while True:
+            msg = yield self.control_queue.get()
+            yield self.process_control(msg)
+
+    @gen.coroutine
+    def process_control(self, msg):
         """dispatch control requests"""
         idents, msg = self.session.feed_identities(msg, copy=False)
         try:
@@ -366,7 +400,12 @@ def dispatch_queue(self):
         Ensures that only one message is processing at a time,
         even when the handler is async
         """
+
         while True:
+            # ensure control stream is flushed before processing shell messages
+            if self.control_stream:
+                self.control_stream.flush()
+            # receive the next message and handle it
             try:
                 yield self.process_one()
             except Exception:
@@ -401,6 +440,7 @@ def start(self):
         self.io_loop.add_callback(self.dispatch_queue)
 
         self.control_stream.on_recv(self.dispatch_control, copy=False)
+        self.debugpy_stream.on_recv(self.dispatch_debugpy, copy=False)
 
         self.shell_stream.on_recv(
             partial(
@@ -442,6 +482,13 @@ def _publish_status(self, status, channel, parent=None):
                           parent=parent or self._parent_header[channel],
                           ident=self._topic('status'),
                           )
+    def _publish_debug_event(self, event):
+        self.session.send(self.iopub_socket,
+                          'debug_event',
+                          event,
+                          parent=self._parent_header['control'],
+                          ident=self._topic('debug_event')
+        )
 
     def set_parent(self, ident, parent, channel='shell'):
         """Set the current parent_header
@@ -693,6 +740,20 @@ def do_is_complete(self, code):
         return {'status' : 'unknown',
                 }
 
+    @gen.coroutine
+    def debug_request(self, stream, ident, parent):
+        content = parent['content']
+
+        reply_content = yield gen.maybe_future(self.do_debug_request(content))
+        reply_content = json_clean(reply_content)
+        reply_msg = self.session.send(stream, 'debug_reply', reply_content,
+                                      parent, ident)
+        self.log.debug("%s", reply_msg)
+
+    @gen.coroutine
+    def do_debug_request(self, msg):
+        return (yield self.debugger.process_request(msg))
+
     #---------------------------------------------------------------------------
     # Engine methods (DEPRECATED)
     #---------------------------------------------------------------------------
diff --git a/ipykernel/kernelspec.py b/ipykernel/kernelspec.py
index f37f98e19..deda6ebec 100644
--- a/ipykernel/kernelspec.py
+++ b/ipykernel/kernelspec.py
@@ -55,6 +55,7 @@ def get_kernel_dict(extra_arguments=None):
         'argv': make_ipkernel_cmd(extra_arguments=extra_arguments),
         'display_name': 'Python %i' % sys.version_info[0],
         'language': 'python',
+        'metadata': { 'debugger': True}
     }
 
 
diff --git a/setup.py b/setup.py
index 106eceaa6..927152151 100644
--- a/setup.py
+++ b/setup.py
@@ -89,7 +89,8 @@ def run(self):
     keywords=['Interactive', 'Interpreter', 'Shell', 'Web'],
     python_requires='>=3.5',
     install_requires=[
-        'ipython>=5.0.0',
+        'debugpy>=1.0.0',
+        'ipython>=7.21.0',
         'traitlets>=4.1.0',
         'jupyter_client',
         'tornado>=4.2',