API Reference

Sans-IO Client

Debug Adapter Protocol (DAP) Client implementation without connection handling.

The client is responsible for sending requests and receiving events from the debug adapter. It is up to the user to handle the connection to the debug adapter, making it more flexible.

The client can be used in a synchronous or asynchronous manner, depending on the user's needs.

Source code in src\dap\client.py

class Client:
    """Debug Adapter Protocol (DAP) Client implementation without connection handling.

    The client is responsible for sending requests and receiving events from the debug adapter.
    It is up to the user to handle the connection to the debug adapter, making it more flexible.

    The client can be used in a synchronous or asynchronous manner, depending on the user's needs.
    """

    def __init__(
        self,
        adapter_id: str,
        client_id: Optional[str] = None,
        client_name: Optional[str] = None,
        locale: Optional[str] = None,
        lines_start_at1: Optional[bool] = None,
        columns_start_at1: Optional[bool] = None,
        path_format: Optional[Literal["path", "uri"] | str] = None,
    ) -> None:
        """Initializes the debug adapter client.

        Args:
            adapter_id: The ID of the debug adapter.
            client_id: The ID of the client.
            client_name: The name of the client.
            locale: The locale of the client.
            lines_start_at1: Whether the lines start at 1.
            columns_start_at1: Whether the columns start at 1.
            path_format: The format of the paths.
        """

        self._seq: int = 1
        self._send_buf = bytearray()
        self._receive_buf = bytearray()
        self._pending_requests: dict[int, Request] = {}

        self.handler = Handler(self)

        self.initialize(
            adapter_id=adapter_id,
            client_id=client_id,
            client_name=client_name,
            locale=locale,
            lines_start_at1=lines_start_at1,
            columns_start_at1=columns_start_at1,
            path_format=path_format,
            supports_variable_type=True,
            supports_variable_paging=True,
            supports_run_in_terminal_request=True,
            supports_memory_references=True,
            supports_progress_reporting=True,
            supports_invalidated_event=True,
            supports_memory_event=True,
            supports_args_can_be_interpreted_by_shell=True,
            supports_start_debugging_request=True,
        )

    def send_request(
        self, command: str, arguments: Optional[dict[str, Any]] = None
    ) -> int:
        """Send a request to the debug adapter.

        This can be useful for sending requests that are not yet implemented in the client or
        for sending custom requests to the debug adapter that are specific to the adapter.

        Args:
            command: The command to send.
            arguments: The arguments to send.

        Returns:
            The sequence number of the request.
        """

        seq = self._seq
        self._seq += 1

        self._send_buf += RequestBuffer(seq, command, arguments)
        self._pending_requests[seq] = Request(
            seq=seq, command=command, arguments=arguments
        )
        return seq

    def receive(self, data: bytes) -> Generator[ResponseBody | EventBody, None, None]:
        """Feed data from the debug adapter to the client.

        Args:
            data: The data to receive.

        Yields:
            The response or event body.
        """

        self._receive_buf += data
        yield from self.handler.handle()

    def send(self) -> bytes:
        """Get the data to send to the debug adapter.

        Returns:
            The data to send.
        """

        send_buf = self._send_buf
        self._send_buf = bytearray()
        return send_buf

    # Requests
    def cancel(
        self, request_id: Optional[int] = None, progress_id: Optional[str] = None
    ) -> int:
        """The cancel request is used by the client in two situations:

        - to indicate that it is no longer interested in the result produced by a specific request issued earlier
        - to cancel a progress sequence.

        Both `progress_id` and `request_id` CAN BE specified in the same request.

        Args:
            request_id: The ID (_seq) of the request to cancel. If missing no request is canceled.
            progress_id: The progress ID of the progress sequence to cancel. If missing no progress is canceled.
        """

        return self.send_request(
            "cancel", {"requestId": request_id, "progressId": progress_id}
        )

    def attach(self, __restart: Optional[Any] = None) -> int:
        """attach to a running process.

        Args:
            __restart: Arbitrary data from the previous, restarted session. \
                The data is sent as the `restart` attribute of the `terminated` event."""

        return self.send_request("attach", {"__restart": __restart})

    def breakpoint_locations(
        self,
        source: Source,
        line: int,
        column: Optional[int] = None,
        end_line: Optional[int] = None,
        end_column: Optional[int] = None,
    ) -> int:
        """Retrieve all possible locations for source breakpoints in a given range.

        Args:
            source: The source location of the breakpoints.
            line: The source line of the breakpoints.
            column: An optional source column of the breakpoints.
            end_line: An optional end line of the range covered by the breakpoint.
            end_column: An optional end column of the range covered by the breakpoint.
        """

        return self.send_request(
            "breakpointLocations",
            {
                "source": source,
                "line": line,
                "column": column,
                "endLine": end_line,
                "endColumn": end_column,
            },
        )

    def completions(
        self,
        text: str,
        column: int,
        line: Optional[int],
        frame_id: Optional[int] = None,
    ) -> int:
        """Returns a list of possible completions for a given caret position and text.

        Args:
            text: One or more source lines. Typically this is the text users have typed into \
                the debug console before they asked for completion.
            column: The position within `text` for which to determine the completion proposals.
            line: A line for which to determine the completion proposals. If missing the \
                first line of the text is assumed.
            frame_id: An optional frameId of the stack frame, if specified returns \
                completions in the scope of this stack frame.
        """

        return self.send_request(
            "completions",
            {
                "frameId": frame_id,
                "text": text,
                "column": column,
                "line": line,
            },
        )

    def configuration_done(self) -> int:
        """This request indicates that the client has finished initialization of the debug adapter."""

        return self.send_request("configurationDone")

    def continue_(self, thread_id: int, single_thread: Optional[bool] = None) -> int:
        """The request resumes execution of all threads.
        If the debug adapter supports single thread execution, setting `single_thread` true resumes only the specified thread.

        Args:
            thread_id: the active thread.
            single_thread: Execute only this thread.
        """

        return self.send_request(
            "continue", {"threadId": thread_id, "singleThread": single_thread}
        )

    def data_breakpoint_info(
        self,
        name: str,
        variables_reference: Optional[int] = None,
        frameId: Optional[int] = None,
        bytes: Optional[int] = None,
        asAddress: Optional[bool] = None,
        mode: Optional[str] = None,
    ) -> int:
        """Retrieve the information of a data breakpoint.

        Args:
            variables_reference: Reference to the variable container if the data breakpoint is requested for \
                a child of the container.
            name: The name of the variable's child to obtain data breakpoint information for.
            frameId: When `name` is an expression, evaluate it in the scope of this stack frame.
            bytes: If specified, a debug adapter should return information for the range of memory extending \
                `bytes` number of bytes from the address or variable specified by `name`. \
                Breakpoints set using the resulting data ID should pause on data access anywhere within that range.
            asAddress: If true, `name` is an address.
            mode: The mode of the desired breakpoint.
        """

        return self.send_request(
            "dataBreakpointInfo",
            {
                "variablesReference": variables_reference,
                "name": name,
                "frameId": frameId,
                "bytes": bytes,
                "asAddress": asAddress,
                "mode": mode,
            },
        )

    def disassemble(
        self,
        memory_reference: str,
        instruction_count: Optional[int] = None,
        offset: Optional[int] = None,
        instruction_offset: Optional[int] = None,
        resolve_symbols: Optional[bool] = None,
    ) -> int:
        """Disassembles code stored at the provided location.

        Args:
            memory_reference: Memory reference to the base location containing the instructions to disassemble.
            instruction_count: The number of instructions to disassemble starting at the specified location and offset.
            offset: The offset (in bytes) of the first instruction to disassemble.
            instruction_offset: The offset (in instructions) of the first instruction to disassemble.
            resolve_symbols: If set to true, the adapter should attempt to resolve memory addresses \
                to function names and line numbers.
        """

        return self.send_request(
            "disassemble",
            {
                "memoryReference": memory_reference,
                "offset": offset,
                "instructionCount": instruction_count,
            },
        )

    def disconnect(
        self,
        restart: Optional[bool] = None,
        terminal_debuggee: Optional[bool] = None,
        suspend_debuggee: Optional[bool] = None,
    ) -> int:
        """Asks the debug adapter to disconnect from the debuggee (thus ending the debug session) and then to shut down.

        In addition, the debug adapter must terminate the debuggee if it was started with the launch request.
        If an attach request was used to connect to the debuggee, then the debug adapter must not terminate the debuggee.

        Args:
            restart: A value of true indicates that this 'disconnect' request is part of a restart sequence.
            terminal_debuggee: Indicates whether the debuggee should be terminated when the debugger is disconnected.
            suspend_debuggee: Indicates whether the debuggee should be allowed to run after the debugger is disconnected.
        """

        return self.send_request(
            "disconnect",
            {
                "restart": restart,
                "terminateDebuggee": terminal_debuggee,
                "suspendDebuggee": suspend_debuggee,
            },
        )

    def evaluate(
        self,
        expression: str,
        frame_id: Optional[int] = None,
        line: Optional[int] = None,
        column: Optional[int] = None,
        source: Optional[Source] = None,
        context: Optional[
            Literal["watch", "repl", "hover", "clipboard", "variables"] | str
        ] = None,
        format: Optional[ValueFormat] = None,
    ) -> int:
        """Evaluate the given expression in the context of topmost stack frame.
        The expression has access to any variables and arguments that are in scope.

        Args:
            expression: The expression to evaluate.
            frame_id: Evaluate the expression in the scope of this stack frame. \
                If not specified, the expression is evaluated in the global scope.
            line: The contextual line where the expression should be evaluated. \
                In the 'hover' context, this should be set to the start of the expression being hovered.
            column: The contextual column where the expression should be evaluated. \
                This may be provided if `line` is also provided.
            source: The contextual source in which the `line` is found. \
                This must be provided if `line` is provided.
            context: The context in which the evaluate request is used.
            format: Specifies details on how to format the result.
        """

        return self.send_request(
            "evaluate",
            {
                "expression": expression,
                "frameId": frame_id,
                "line": line,
                "column": column,
                "source": source,
                "context": context,
                "format": format,
            },
        )

    def exception_info(self, thread_id: int) -> int:
        """Retrieves the details of the exception that caused this event to be raised.

        Args:
            thread_id: Thread for which exception information should be retrieved.
        """

        return self.send_request("exceptionInfo", {"threadId": thread_id})

    def goto(self, thread_id: int, target_id: str) -> int:
        """The request sets the location where the debuggee will continue to run.

        Args:
            thread_id: The thread to continue.
            target_id: The location where the debuggee will continue to run.
        """

        return self.send_request("goto", {"threadId": thread_id, "targetId": target_id})

    def goto_targets(
        self, source: Source, line: int, column: Optional[int] = None
    ) -> int:
        """Retrieve possible goto targets for the specified location.

        Args:
            source: The source location for which the goto targets are determined.
            line: The line for which the goto targets are determined.
            column: An optional column for which the goto targets are determined.
        """

        return self.send_request(
            "gotoTargets", {"source": source, "line": line, "column": column}
        )

    def initialize(
        self,
        adapter_id: str,
        client_id: Optional[str] = None,
        client_name: Optional[str] = None,
        locale: Optional[str] = None,
        lines_start_at1: Optional[bool] = None,
        columns_start_at1: Optional[bool] = None,
        path_format: Optional[Literal["path", "uri"] | str] = None,
        supports_variable_type: Optional[bool] = None,
        supports_variable_paging: Optional[bool] = None,
        supports_run_in_terminal_request: Optional[bool] = None,
        supports_memory_references: Optional[bool] = None,
        supports_progress_reporting: Optional[bool] = None,
        supports_invalidated_event: Optional[bool] = None,
        supports_memory_event: Optional[bool] = None,
        supports_args_can_be_interpreted_by_shell: Optional[bool] = None,
        supports_start_debugging_request: Optional[bool] = None,
    ) -> int:
        """Initializes the debug adapter with the client capabilities."""

        return self.send_request(
            "initialize",
            {
                "adapterID": adapter_id,
                "clientID": client_id,
                "clientName": client_name,
                "locale": locale,
                "linesStartAt1": lines_start_at1,
                "columnsStartAt1": columns_start_at1,
                "pathFormat": path_format,
                "supportsVariableType": supports_variable_type,
                "supportsVariablePaging": supports_variable_paging,
                "supportsRunInTerminalRequest": supports_run_in_terminal_request,
                "supportsMemoryReferences": supports_memory_references,
                "supportsProgressReporting": supports_progress_reporting,
                "supportsInvalidatedEvent": supports_invalidated_event,
                "supportsMemoryEvent": supports_memory_event,
                "supportsArgsCanBeInterpretedByShell": supports_args_can_be_interpreted_by_shell,
                "supportsStartDebuggingRequest": supports_start_debugging_request,
            },
        )

    def launch(
        self,
        no_debug: Optional[bool] = None,
        __restart: Optional[Any] = None,
    ) -> int:
        """The launch request is used to start the debuggee with or without debugging enabled.

        Args:
            no_debug: Set to true if the launch request is used to just start the debuggee \
                for the purpose of collecting output. The debuggee is not supposed to stop at breakpoints.
            __restart: Arbitrary data from the previous, restarted session. \
                The data is sent as the `restart` attribute of the `terminated` event.
        """

        return self.send_request(
            "launch",
            {"noDebug": no_debug, "__restart": __restart},
        )

    def loaded_sources(self) -> int:
        """Retrieves the set of all sources currently loaded by the debugged process."""

        return self.send_request("loadedSources")

    def modules(
        self, start_module: Optional[int] = None, module_count: Optional[int] = None
    ) -> int:
        """Modules can be retrieved from the debug adapter with this request which can either
        return all modules or a range of modules to support paging.

        Args:
            start_module: The 0-based index of the first module to return; if omitted modules start at 0.
            module_count: The number of modules to return. If moduleCount is not specified or 0, \
                all modules are returned.
        """

        return self.send_request(
            "modules", {"startModule": start_module, "moduleCount": module_count}
        )

    def next(
        self,
        thread_id: int,
        single_thread: Optional[bool] = None,
        granularity: Optional[str] = None,
    ) -> int:
        """The request steps through the program.

        Args:
            thread_id: Specifies the thread for which to resume execution for one step.
            single_thread: If this is true, all other suspended threads are not resumed.
            granularity: The granularity of the step, assumed to be 'statement' if not specified.
        """

        return self.send_request(
            "next",
            {
                "threadId": thread_id,
                "singleThread": single_thread,
                "granularity": granularity,
            },
        )

    def pause(self, thread_id: int) -> int:
        """Suspends the debuggee.

        Args:
            thread_id: The thread to pause.
        """

        return self.send_request("pause", {"threadId": thread_id})

    def read_memory(
        self, memory_reference: str, count: int, offset: Optional[int] = None
    ) -> int:
        """Reads memory from the debuggee.

        Args:
            memory_reference: The memory reference to the base location from which to read memory.
            count: The number of bytes to read at the specified location and offset.
            offset: The offset (in bytes) of the first byte to read.
        """

        return self.send_request(
            "readMemory",
            {"memoryReference": memory_reference, "offset": offset, "count": count},
        )

    def restart(
        self,
        arguments: Optional[LaunchRequestArguments | AttachRequestArguments] = None,
    ) -> int:
        """Restarts a debug session.

        Args:
            arguments: Use either arguments for the 'launch' or 'attach' request.
        """

        return self.send_request("restart", arguments)

    def restart_frame(self, frame_id: int) -> int:
        """Restart the stack frame identified by the given frame ID.
        The frame ID must have been obtained in the current suspended state.

        Args:
            frame_id: The frame to restart.
        """

        return self.send_request("restartFrame", {"frameId": frame_id})

    def reverse_continue(
        self, thread_id: int, single_thread: Optional[bool] = None
    ) -> int:
        """The request starts the debuggee to run backward.

        Args:
            thread_id: ID of the active thread.
            single_thread: If true, backward execution is limited to the specified thread.
        """

        return self.send_request(
            "reverseContinue",
            {"threadId": thread_id, "singleThread": single_thread},
        )

    def scopes(self, frame_id: int) -> int:
        """The request returns the variable scopes for a given stack frame.

        Args:
            frame_id: Retrieve the scopes for this stackframe.
        """

        return self.send_request("scopes", {"frameId": frame_id})

    def set_breakpoints(
        self,
        source: Source,
        breakpoints: List[SourceBreakpoint],
        lines: Optional[List[int]] = None,
        source_modified: Optional[bool] = None,
    ) -> int:
        """Sets multiple breakpoints for a single source and clears all previous breakpoints in that source.

        Args:
            source: The source location of the breakpoints.
            breakpoints: The code locations of the breakpoints.
            lines: Deprecated: The code locations of the breakpoints.
            source_modified: A value of true indicates that the underlying source has been modified \
                which results in new breakpoint locations.
        """

        return self.send_request(
            "setBreakpoints",
            {
                "source": source,
                "breakpoints": breakpoints,
                "lines": lines,
                "sourceModified": source_modified,
            },
        )

    def set_data_breakpoints(self, breakpoints: List[DataBreakpoint]) -> int:
        """Replaces all existing data breakpoints with new data breakpoints.

        Args:
            breakpoints: The data breakpoints to set.
        """

        return self.send_request("setDataBreakpoints", {"breakpoints": breakpoints})

    def set_exception_breakpoints(
        self,
        filters: List[str],
        filter_options: Optional[List[ExceptionFilterOptions]],
        exception_options: Optional[List[ExceptionOptions]],
    ) -> int:
        """The request configures the debugger's response to thrown exceptions.

        Each of the filters, filterOptions, and exceptionOptions in the request are independent configurations
        to a debug adapter indicating a kind of exception to catch. An exception thrown in a program should result
        in a stopped event from the debug adapter (with reason exception) if any of the configured filters match.

        Args:
            filters: Set of exception filters specified by their ID.
            filter_options: An array of ExceptionFilterOptions. The set of all possible exception filters \
                is defined by the `exceptionBreakpointFilters` capability.
            exception_options: An array of ExceptionOptions. Configuration options for selected exceptions.
        """

        return self.send_request(
            "setExceptionBreakpoints",
            {
                "filters": filters,
                "filterOptions": filter_options,
                "exceptionOptions": exception_options,
            },
        )

    def set_expression(
        self,
        expression: str,
        value: str,
        frame_id: Optional[int] = None,
        format: Optional[ValueFormat] = None,
    ) -> int:
        """Evaluates the given value expression and assigns it to the expression which must be a modifiable l-value.

        The expressions have access to any variables and arguments that are in scope of the specified frame.

        Args:
            expression: The l-value expression to assign the result to.
            value: The value expression to assign to the l-value expression.
            frame_id: Evaluate the expressions in the scope of this stack frame. \
                If not specified, the expressions are evaluated in the global scope.
            format: Specifies details on how to format the result.
        """

        return self.send_request(
            "setExpression",
            {
                "expression": expression,
                "value": value,
                "frameId": frame_id,
                "format": format,
            },
        )

    def set_function_breakpoints(
        self, breakpoints: List[FunctionBreakpoint] = []
    ) -> int:
        """Replaces all existing function breakpoints with new function breakpoints.

        To clear all function breakpoints, call this without arguments.
        When a function breakpoint is hit, a stopped event (with reason function breakpoint) is generated.

        Args:
            breakpoints: The function breakpoints to set.
        """

        return self.send_request("setFunctionBreakpoints", {"breakpoints": breakpoints})

    def set_instruction_breakpoints(
        self, breakpoints: List[InstructionBreakpoint]
    ) -> int:
        """Replaces all existing instruction breakpoints. Typically, instruction breakpoints would be set from a disassembly window.

        To clear all instruction breakpoints, specify an empty array.
        When an instruction breakpoint is hit, a stopped event (with reason instruction breakpoint) is generated.

        Args:
            breakpoints: The instruction breakpoints to set.
        """

        return self.send_request(
            "setInstructionBreakpoints", {"breakpoints": breakpoints}
        )

    def set_variable(
        self,
        variables_reference: int,
        name: str,
        value: str,
        format: Optional[ValueFormat] = None,
    ) -> int:
        """Set the variable with the given name in the variable container to a new value.

        Args:
            variables_reference: The reference of the variable container.
            name: The name of the variable to set.
            value: The value to set.
            format: Specifies details on how to format the response value.
        """

        return self.send_request(
            "setVariable",
            {
                "variablesReference": variables_reference,
                "name": name,
                "value": value,
                "format": format,
            },
        )

    def source(self, source_reference: int, source: Optional[Source] = None) -> int:
        """The request retrieves the source code for a given source reference.

        Args:
            source_reference: The reference to the source. This is the same as `source.sourceReference`.
            source: Specifies the source content to load. Either `source.path` or `source.sourceReference` must be specified.
        """

        return self.send_request(
            "source", {"sourceReference": source_reference, "source": source}
        )

    def stack_trace(
        self,
        thread_id: Optional[int] = None,
        start_frame: Optional[int] = None,
        levels: Optional[int] = None,
        format: Optional[StackFrameFormat] = None,
    ) -> int:
        """The request returns a stack trace from the current execution state.

        Request all stack frames by omitting the startFrame and levels arguments.

        Args:
            thread_id: Retrieve the stacktrace for this thread.
            start_frame: The index of the first frame to return; if omitted frames start at 0.
            levels: The maximum number of frames to return. If levels is not specified or 0, all frames are returned.
            format: Specifies details on how to format the stack frames.
        """

        return self.send_request(
            "stackTrace",
            {
                "threadId": thread_id,
                "startFrame": start_frame,
                "levels": levels,
                "format": format,
            },
        )

    def step_back(
        self,
        thread_id: int,
        single_thread: Optional[bool] = None,
        granularity: Optional[SteppingGranularity] = None,
    ) -> int:
        """The request executes one backward step (in the given granularity) for the specified thread
        and allows all other threads to run backward freely by resuming them.

        If the debug adapter supports single thread execution (see capability `supportsSingleThreadExecutionRequests`),
        setting the singleThread argument to true prevents other suspended threads from resuming.
        The debug adapter first sends the response and then a stopped event (with reason step) after the step has completed.

        Args:
            thread_id: ID of the active thread.
            single_thread: If true, backward execution is limited to the specified thread.
            granularity: The granularity of the step, assumed to be 'statement' if not specified.
        """

        return self.send_request(
            "stepBack",
            {
                "threadId": thread_id,
                "singleThread": single_thread,
                "granularity": granularity,
            },
        )

    def step_in(
        self,
        thread_id: int,
        single_thread: Optional[bool] = None,
        target_id: Optional[int] = None,
        granularity: Optional[SteppingGranularity] = None,
    ) -> int:
        """The request resumes the given thread to step into a function/method and allows all other threads to run freely by resuming them.

        If the debug adapter supports single thread execution (see capability `supportsSingleThreadExecutionRequests`),
        setting the singleThread argument to true prevents other suspended threads from resuming.

        If the request cannot step into a target, stepIn behaves like the next request.
        The debug adapter first sends the response and then a stopped event (with reason step) after the step has completed.

        If there are multiple function/method calls (or other targets) on the source line,
        the argument targetId can be used to control into which target the stepIn should occur.

        Args:
            thread_id: ID of the active thread.
            single_thread: If true, stepIn is limited to the specified thread.
            target_id: The stepIn target for this step.
            granularity: The granularity of the step, assumed to be 'statement' if not specified.
        """

        return self.send_request(
            "stepIn",
            {
                "threadId": thread_id,
                "singleThread": single_thread,
                "targetId": target_id,
                "granularity": granularity,
            },
        )

    def step_in_targets(self, frame_id: int) -> int:
        """The request retrieves the possible stepIn targets for the specified stack frame.
        These targets can be used in the stepIn request.

        Args:
            frame_id: The stack frame for which to retrieve the possible stepIn targets.
        """

        return self.send_request("stepInTargets", {"frameId": frame_id})

    def step_out(
        self,
        thread_id: int,
        single_thread: Optional[bool] = None,
        granularity: Optional[SteppingGranularity] = None,
    ) -> int:
        """The request resumes the given thread to step out of the current function/method and allows all other threads to run freely by resuming them.

        If the debug adapter supports single thread execution (see capability `supportsSingleThreadExecutionRequests`),
        setting the singleThread argument to true prevents other suspended threads from resuming.

        The debug adapter first sends the response and then a stopped event (with reason step) after the step has completed.

        Args:
            thread_id: ID of the active thread.
            single_thread: If true, stepOut is limited to the specified thread.
            granularity: The granularity of the step, assumed to be 'statement' if not specified.
        """

        return self.send_request(
            "stepOut",
            {
                "threadId": thread_id,
                "singleThread": single_thread,
                "granularity": granularity,
            },
        )

    def terminate(self, restart: Optional[bool] = None) -> int:
        """The terminate request is sent from the client to the debug adapter in order to shut down the debuggee gracefully.

        Typically a debug adapter implements terminate by sending a software signal which the debuggee intercepts in order
        to clean things up properly before terminating itself.

        Please note that this request does not directly affect the state of the debug session: if the debuggee decides to
        veto the graceful shutdown for any reason by not terminating itself, then the debug session just continues.

        Args:
            restart: A value of true indicates that this 'terminate' request is part of a restart sequence.
        """

        return self.send_request("terminate", {"restart": restart})

    def terminate_threads(self, thread_ids: List[int]) -> int:
        """The request terminates the threads with the given ids.

        Args:
            thread_ids: The threads to terminate.
        """

        return self.send_request("terminateThreads", {"threadIds": thread_ids})

    def threads(self) -> int:
        """The request retrieves a list of all threads."""

        return self.send_request("threads")

    def variables(
        self,
        variables_reference: int,
        filter: Optional[Literal["indexed", "named"]] | str = None,
        start: Optional[int] = None,
        count: Optional[int] = None,
        format: Optional[ValueFormat] = None,
    ) -> int:
        """Retrieves all child variables for the given variable reference.

        A filter can be used to limit the fetched children to either named or indexed children.

        Args:
            variables_reference: The variable for which to retrieve its children.
            filter: Filter to limit the child variables to either named or indexed. If not specified, both types are fetched.
            start: The index of the first variable to return; if omitted variables start at 0.
            count: The number of variables to return. If not passed or 0, all variables are returned.
            format: Specifies details on how to format the response value.
        """

        return self.send_request(
            "variables",
            {
                "variablesReference": variables_reference,
                "filter": filter,
                "start": start,
                "count": count,
                "format": format,
            },
        )

    def write_memory(
        self,
        memory_reference: str,
        data: str,
        offset: Optional[int] = None,
        allow_partial: Optional[bool] = None,
    ) -> int:
        """Writes bytes to memory at the provided location.

        Args:
            memory_reference: The memory reference to the base location to write memory.
            data: Bytes to write, encoded using base64.
            offset: The offset (in bytes) of the first byte to write. Can be negative.
            allow_partial: Property to control partial writes. If true, the debug adapter should \
                attempt to write memory even if the entire memory region is not writable.
        """

        return self.send_request(
            "writeMemory",
            {
                "memoryReference": memory_reference,
                "offset": offset,
                "data": data,
                "allowPartial": allow_partial,
            },
        )

__init__(adapter_id, client_id=None, client_name=None, locale=None, lines_start_at1=None, columns_start_at1=None, path_format=None)

Initializes the debug adapter client.

Parameters:

Name	Type	Description	Default
adapter_id	str	The ID of the debug adapter.	required
client_id	Optional[str]	The ID of the client.	None
client_name	Optional[str]	The name of the client.	None
locale	Optional[str]	The locale of the client.	None
lines_start_at1	Optional[bool]	Whether the lines start at 1.	None
columns_start_at1	Optional[bool]	Whether the columns start at 1.	None
path_format	Optional[Literal[path, uri] | str]	The format of the paths.	None

Source code in src\dap\client.py

def __init__(
    self,
    adapter_id: str,
    client_id: Optional[str] = None,
    client_name: Optional[str] = None,
    locale: Optional[str] = None,
    lines_start_at1: Optional[bool] = None,
    columns_start_at1: Optional[bool] = None,
    path_format: Optional[Literal["path", "uri"] | str] = None,
) -> None:
    """Initializes the debug adapter client.

    Args:
        adapter_id: The ID of the debug adapter.
        client_id: The ID of the client.
        client_name: The name of the client.
        locale: The locale of the client.
        lines_start_at1: Whether the lines start at 1.
        columns_start_at1: Whether the columns start at 1.
        path_format: The format of the paths.
    """

    self._seq: int = 1
    self._send_buf = bytearray()
    self._receive_buf = bytearray()
    self._pending_requests: dict[int, Request] = {}

    self.handler = Handler(self)

    self.initialize(
        adapter_id=adapter_id,
        client_id=client_id,
        client_name=client_name,
        locale=locale,
        lines_start_at1=lines_start_at1,
        columns_start_at1=columns_start_at1,
        path_format=path_format,
        supports_variable_type=True,
        supports_variable_paging=True,
        supports_run_in_terminal_request=True,
        supports_memory_references=True,
        supports_progress_reporting=True,
        supports_invalidated_event=True,
        supports_memory_event=True,
        supports_args_can_be_interpreted_by_shell=True,
        supports_start_debugging_request=True,
    )

attach(__restart=None)

attach to a running process.

Parameters:

Name	Type	Description	Default
__restart	Optional[Any]	Arbitrary data from the previous, restarted session. The data is sent as the restart attribute of the terminated event.	None

Source code in src\dap\client.py

def attach(self, __restart: Optional[Any] = None) -> int:
    """attach to a running process.

    Args:
        __restart: Arbitrary data from the previous, restarted session. \
            The data is sent as the `restart` attribute of the `terminated` event."""

    return self.send_request("attach", {"__restart": __restart})

breakpoint_locations(source, line, column=None, end_line=None, end_column=None)

Retrieve all possible locations for source breakpoints in a given range.

Parameters:

Name	Type	Description	Default
source	Source	The source location of the breakpoints.	required
line	int	The source line of the breakpoints.	required
column	Optional[int]	An optional source column of the breakpoints.	None
end_line	Optional[int]	An optional end line of the range covered by the breakpoint.	None
end_column	Optional[int]	An optional end column of the range covered by the breakpoint.	None

Source code in src\dap\client.py

def breakpoint_locations(
    self,
    source: Source,
    line: int,
    column: Optional[int] = None,
    end_line: Optional[int] = None,
    end_column: Optional[int] = None,
) -> int:
    """Retrieve all possible locations for source breakpoints in a given range.

    Args:
        source: The source location of the breakpoints.
        line: The source line of the breakpoints.
        column: An optional source column of the breakpoints.
        end_line: An optional end line of the range covered by the breakpoint.
        end_column: An optional end column of the range covered by the breakpoint.
    """

    return self.send_request(
        "breakpointLocations",
        {
            "source": source,
            "line": line,
            "column": column,
            "endLine": end_line,
            "endColumn": end_column,
        },
    )

cancel(request_id=None, progress_id=None)

The cancel request is used by the client in two situations:

• to indicate that it is no longer interested in the result produced by a specific request issued earlier
• to cancel a progress sequence.

Both progress_id and request_id CAN BE specified in the same request.

Parameters:

Name	Type	Description	Default
request_id	Optional[int]	The ID (_seq) of the request to cancel. If missing no request is canceled.	None
progress_id	Optional[str]	The progress ID of the progress sequence to cancel. If missing no progress is canceled.	None

Source code in src\dap\client.py

def cancel(
    self, request_id: Optional[int] = None, progress_id: Optional[str] = None
) -> int:
    """The cancel request is used by the client in two situations:

    - to indicate that it is no longer interested in the result produced by a specific request issued earlier
    - to cancel a progress sequence.

    Both `progress_id` and `request_id` CAN BE specified in the same request.

    Args:
        request_id: The ID (_seq) of the request to cancel. If missing no request is canceled.
        progress_id: The progress ID of the progress sequence to cancel. If missing no progress is canceled.
    """

    return self.send_request(
        "cancel", {"requestId": request_id, "progressId": progress_id}
    )

completions(text, column, line, frame_id=None)

Returns a list of possible completions for a given caret position and text.

Parameters:

Name	Type	Description	Default
text	str	One or more source lines. Typically this is the text users have typed into the debug console before they asked for completion.	required
column	int	The position within text for which to determine the completion proposals.	required
line	Optional[int]	A line for which to determine the completion proposals. If missing the first line of the text is assumed.	required
frame_id	Optional[int]	An optional frameId of the stack frame, if specified returns completions in the scope of this stack frame.	None

Source code in src\dap\client.py

def completions(
    self,
    text: str,
    column: int,
    line: Optional[int],
    frame_id: Optional[int] = None,
) -> int:
    """Returns a list of possible completions for a given caret position and text.

    Args:
        text: One or more source lines. Typically this is the text users have typed into \
            the debug console before they asked for completion.
        column: The position within `text` for which to determine the completion proposals.
        line: A line for which to determine the completion proposals. If missing the \
            first line of the text is assumed.
        frame_id: An optional frameId of the stack frame, if specified returns \
            completions in the scope of this stack frame.
    """

    return self.send_request(
        "completions",
        {
            "frameId": frame_id,
            "text": text,
            "column": column,
            "line": line,
        },
    )

configuration_done()

This request indicates that the client has finished initialization of the debug adapter.

Source code in src\dap\client.py

def configuration_done(self) -> int:
    """This request indicates that the client has finished initialization of the debug adapter."""

    return self.send_request("configurationDone")

continue_(thread_id, single_thread=None)

The request resumes execution of all threads. If the debug adapter supports single thread execution, setting single_thread true resumes only the specified thread.

Parameters:

Name	Type	Description	Default
thread_id	int	the active thread.	required
single_thread	Optional[bool]	Execute only this thread.	None

Source code in src\dap\client.py

def continue_(self, thread_id: int, single_thread: Optional[bool] = None) -> int:
    """The request resumes execution of all threads.
    If the debug adapter supports single thread execution, setting `single_thread` true resumes only the specified thread.

    Args:
        thread_id: the active thread.
        single_thread: Execute only this thread.
    """

    return self.send_request(
        "continue", {"threadId": thread_id, "singleThread": single_thread}
    )

data_breakpoint_info(name, variables_reference=None, frameId=None, bytes=None, asAddress=None, mode=None)

Retrieve the information of a data breakpoint.

Parameters:

Name	Type	Description	Default
variables_reference	Optional[int]	Reference to the variable container if the data breakpoint is requested for a child of the container.	None
name	str	The name of the variable's child to obtain data breakpoint information for.	required
frameId	Optional[int]	When name is an expression, evaluate it in the scope of this stack frame.	None
bytes	Optional[int]	If specified, a debug adapter should return information for the range of memory extending bytes number of bytes from the address or variable specified by name. Breakpoints set using the resulting data ID should pause on data access anywhere within that range.	None
asAddress	Optional[bool]	If true, name is an address.	None
mode	Optional[str]	The mode of the desired breakpoint.	None

Source code in src\dap\client.py

def data_breakpoint_info(
    self,
    name: str,
    variables_reference: Optional[int] = None,
    frameId: Optional[int] = None,
    bytes: Optional[int] = None,
    asAddress: Optional[bool] = None,
    mode: Optional[str] = None,
) -> int:
    """Retrieve the information of a data breakpoint.

    Args:
        variables_reference: Reference to the variable container if the data breakpoint is requested for \
            a child of the container.
        name: The name of the variable's child to obtain data breakpoint information for.
        frameId: When `name` is an expression, evaluate it in the scope of this stack frame.
        bytes: If specified, a debug adapter should return information for the range of memory extending \
            `bytes` number of bytes from the address or variable specified by `name`. \
            Breakpoints set using the resulting data ID should pause on data access anywhere within that range.
        asAddress: If true, `name` is an address.
        mode: The mode of the desired breakpoint.
    """

    return self.send_request(
        "dataBreakpointInfo",
        {
            "variablesReference": variables_reference,
            "name": name,
            "frameId": frameId,
            "bytes": bytes,
            "asAddress": asAddress,
            "mode": mode,
        },
    )

disassemble(memory_reference, instruction_count=None, offset=None, instruction_offset=None, resolve_symbols=None)

Disassembles code stored at the provided location.

Parameters:

Name	Type	Description	Default
memory_reference	str	Memory reference to the base location containing the instructions to disassemble.	required
instruction_count	Optional[int]	The number of instructions to disassemble starting at the specified location and offset.	None
offset	Optional[int]	The offset (in bytes) of the first instruction to disassemble.	None
instruction_offset	Optional[int]	The offset (in instructions) of the first instruction to disassemble.	None
resolve_symbols	Optional[bool]	If set to true, the adapter should attempt to resolve memory addresses to function names and line numbers.	None

Source code in src\dap\client.py

def disassemble(
    self,
    memory_reference: str,
    instruction_count: Optional[int] = None,
    offset: Optional[int] = None,
    instruction_offset: Optional[int] = None,
    resolve_symbols: Optional[bool] = None,
) -> int:
    """Disassembles code stored at the provided location.

    Args:
        memory_reference: Memory reference to the base location containing the instructions to disassemble.
        instruction_count: The number of instructions to disassemble starting at the specified location and offset.
        offset: The offset (in bytes) of the first instruction to disassemble.
        instruction_offset: The offset (in instructions) of the first instruction to disassemble.
        resolve_symbols: If set to true, the adapter should attempt to resolve memory addresses \
            to function names and line numbers.
    """

    return self.send_request(
        "disassemble",
        {
            "memoryReference": memory_reference,
            "offset": offset,
            "instructionCount": instruction_count,
        },
    )

disconnect(restart=None, terminal_debuggee=None, suspend_debuggee=None)

Asks the debug adapter to disconnect from the debuggee (thus ending the debug session) and then to shut down.

In addition, the debug adapter must terminate the debuggee if it was started with the launch request. If an attach request was used to connect to the debuggee, then the debug adapter must not terminate the debuggee.

Parameters:

Name	Type	Description	Default
restart	Optional[bool]	A value of true indicates that this 'disconnect' request is part of a restart sequence.	None
terminal_debuggee	Optional[bool]	Indicates whether the debuggee should be terminated when the debugger is disconnected.	None
suspend_debuggee	Optional[bool]	Indicates whether the debuggee should be allowed to run after the debugger is disconnected.	None

Source code in src\dap\client.py

def disconnect(
    self,
    restart: Optional[bool] = None,
    terminal_debuggee: Optional[bool] = None,
    suspend_debuggee: Optional[bool] = None,
) -> int:
    """Asks the debug adapter to disconnect from the debuggee (thus ending the debug session) and then to shut down.

    In addition, the debug adapter must terminate the debuggee if it was started with the launch request.
    If an attach request was used to connect to the debuggee, then the debug adapter must not terminate the debuggee.

    Args:
        restart: A value of true indicates that this 'disconnect' request is part of a restart sequence.
        terminal_debuggee: Indicates whether the debuggee should be terminated when the debugger is disconnected.
        suspend_debuggee: Indicates whether the debuggee should be allowed to run after the debugger is disconnected.
    """

    return self.send_request(
        "disconnect",
        {
            "restart": restart,
            "terminateDebuggee": terminal_debuggee,
            "suspendDebuggee": suspend_debuggee,
        },
    )

evaluate(expression, frame_id=None, line=None, column=None, source=None, context=None, format=None)

Evaluate the given expression in the context of topmost stack frame. The expression has access to any variables and arguments that are in scope.

Parameters:

Name	Type	Description	Default
expression	str	The expression to evaluate.	required
frame_id	Optional[int]	Evaluate the expression in the scope of this stack frame. If not specified, the expression is evaluated in the global scope.	None
line	Optional[int]	The contextual line where the expression should be evaluated. In the 'hover' context, this should be set to the start of the expression being hovered.	None
column	Optional[int]	The contextual column where the expression should be evaluated. This may be provided if line is also provided.	None
source	Optional[Source]	The contextual source in which the line is found. This must be provided if line is provided.	None
context	Optional[Literal[watch, repl, hover, clipboard, variables] | str]	The context in which the evaluate request is used.	None
format	Optional[ValueFormat]	Specifies details on how to format the result.	None

Source code in src\dap\client.py

def evaluate(
    self,
    expression: str,
    frame_id: Optional[int] = None,
    line: Optional[int] = None,
    column: Optional[int] = None,
    source: Optional[Source] = None,
    context: Optional[
        Literal["watch", "repl", "hover", "clipboard", "variables"] | str
    ] = None,
    format: Optional[ValueFormat] = None,
) -> int:
    """Evaluate the given expression in the context of topmost stack frame.
    The expression has access to any variables and arguments that are in scope.

    Args:
        expression: The expression to evaluate.
        frame_id: Evaluate the expression in the scope of this stack frame. \
            If not specified, the expression is evaluated in the global scope.
        line: The contextual line where the expression should be evaluated. \
            In the 'hover' context, this should be set to the start of the expression being hovered.
        column: The contextual column where the expression should be evaluated. \
            This may be provided if `line` is also provided.
        source: The contextual source in which the `line` is found. \
            This must be provided if `line` is provided.
        context: The context in which the evaluate request is used.
        format: Specifies details on how to format the result.
    """

    return self.send_request(
        "evaluate",
        {
            "expression": expression,
            "frameId": frame_id,
            "line": line,
            "column": column,
            "source": source,
            "context": context,
            "format": format,
        },
    )

exception_info(thread_id)

Retrieves the details of the exception that caused this event to be raised.

Parameters:

Name	Type	Description	Default
thread_id	int	Thread for which exception information should be retrieved.	required

Source code in src\dap\client.py

def exception_info(self, thread_id: int) -> int:
    """Retrieves the details of the exception that caused this event to be raised.

    Args:
        thread_id: Thread for which exception information should be retrieved.
    """

    return self.send_request("exceptionInfo", {"threadId": thread_id})

goto(thread_id, target_id)

The request sets the location where the debuggee will continue to run.

Parameters:

Name	Type	Description	Default
thread_id	int	The thread to continue.	required
target_id	str	The location where the debuggee will continue to run.	required

Source code in src\dap\client.py

def goto(self, thread_id: int, target_id: str) -> int:
    """The request sets the location where the debuggee will continue to run.

    Args:
        thread_id: The thread to continue.
        target_id: The location where the debuggee will continue to run.
    """

    return self.send_request("goto", {"threadId": thread_id, "targetId": target_id})

goto_targets(source, line, column=None)

Retrieve possible goto targets for the specified location.

Parameters:

Name	Type	Description	Default
source	Source	The source location for which the goto targets are determined.	required
line	int	The line for which the goto targets are determined.	required
column	Optional[int]	An optional column for which the goto targets are determined.	None

Source code in src\dap\client.py

def goto_targets(
    self, source: Source, line: int, column: Optional[int] = None
) -> int:
    """Retrieve possible goto targets for the specified location.

    Args:
        source: The source location for which the goto targets are determined.
        line: The line for which the goto targets are determined.
        column: An optional column for which the goto targets are determined.
    """

    return self.send_request(
        "gotoTargets", {"source": source, "line": line, "column": column}
    )

initialize(adapter_id, client_id=None, client_name=None, locale=None, lines_start_at1=None, columns_start_at1=None, path_format=None, supports_variable_type=None, supports_variable_paging=None, supports_run_in_terminal_request=None, supports_memory_references=None, supports_progress_reporting=None, supports_invalidated_event=None, supports_memory_event=None, supports_args_can_be_interpreted_by_shell=None, supports_start_debugging_request=None)

Initializes the debug adapter with the client capabilities.

Source code in src\dap\client.py

def initialize(
    self,
    adapter_id: str,
    client_id: Optional[str] = None,
    client_name: Optional[str] = None,
    locale: Optional[str] = None,
    lines_start_at1: Optional[bool] = None,
    columns_start_at1: Optional[bool] = None,
    path_format: Optional[Literal["path", "uri"] | str] = None,
    supports_variable_type: Optional[bool] = None,
    supports_variable_paging: Optional[bool] = None,
    supports_run_in_terminal_request: Optional[bool] = None,
    supports_memory_references: Optional[bool] = None,
    supports_progress_reporting: Optional[bool] = None,
    supports_invalidated_event: Optional[bool] = None,
    supports_memory_event: Optional[bool] = None,
    supports_args_can_be_interpreted_by_shell: Optional[bool] = None,
    supports_start_debugging_request: Optional[bool] = None,
) -> int:
    """Initializes the debug adapter with the client capabilities."""

    return self.send_request(
        "initialize",
        {
            "adapterID": adapter_id,
            "clientID": client_id,
            "clientName": client_name,
            "locale": locale,
            "linesStartAt1": lines_start_at1,
            "columnsStartAt1": columns_start_at1,
            "pathFormat": path_format,
            "supportsVariableType": supports_variable_type,
            "supportsVariablePaging": supports_variable_paging,
            "supportsRunInTerminalRequest": supports_run_in_terminal_request,
            "supportsMemoryReferences": supports_memory_references,
            "supportsProgressReporting": supports_progress_reporting,
            "supportsInvalidatedEvent": supports_invalidated_event,
            "supportsMemoryEvent": supports_memory_event,
            "supportsArgsCanBeInterpretedByShell": supports_args_can_be_interpreted_by_shell,
            "supportsStartDebuggingRequest": supports_start_debugging_request,
        },
    )

launch(no_debug=None, __restart=None)

The launch request is used to start the debuggee with or without debugging enabled.

Parameters:

Name	Type	Description	Default
no_debug	Optional[bool]	Set to true if the launch request is used to just start the debuggee for the purpose of collecting output. The debuggee is not supposed to stop at breakpoints.	None
__restart	Optional[Any]	Arbitrary data from the previous, restarted session. The data is sent as the restart attribute of the terminated event.	None

Source code in src\dap\client.py

def launch(
    self,
    no_debug: Optional[bool] = None,
    __restart: Optional[Any] = None,
) -> int:
    """The launch request is used to start the debuggee with or without debugging enabled.

    Args:
        no_debug: Set to true if the launch request is used to just start the debuggee \
            for the purpose of collecting output. The debuggee is not supposed to stop at breakpoints.
        __restart: Arbitrary data from the previous, restarted session. \
            The data is sent as the `restart` attribute of the `terminated` event.
    """

    return self.send_request(
        "launch",
        {"noDebug": no_debug, "__restart": __restart},
    )

loaded_sources()

Retrieves the set of all sources currently loaded by the debugged process.

Source code in src\dap\client.py

def loaded_sources(self) -> int:
    """Retrieves the set of all sources currently loaded by the debugged process."""

    return self.send_request("loadedSources")

modules(start_module=None, module_count=None)

Modules can be retrieved from the debug adapter with this request which can either return all modules or a range of modules to support paging.

Parameters:

Name	Type	Description	Default
start_module	Optional[int]	The 0-based index of the first module to return; if omitted modules start at 0.	None
module_count	Optional[int]	The number of modules to return. If moduleCount is not specified or 0, all modules are returned.	None

Source code in src\dap\client.py

def modules(
    self, start_module: Optional[int] = None, module_count: Optional[int] = None
) -> int:
    """Modules can be retrieved from the debug adapter with this request which can either
    return all modules or a range of modules to support paging.

    Args:
        start_module: The 0-based index of the first module to return; if omitted modules start at 0.
        module_count: The number of modules to return. If moduleCount is not specified or 0, \
            all modules are returned.
    """

    return self.send_request(
        "modules", {"startModule": start_module, "moduleCount": module_count}
    )

next(thread_id, single_thread=None, granularity=None)

The request steps through the program.

Parameters:

Name	Type	Description	Default
thread_id	int	Specifies the thread for which to resume execution for one step.	required
single_thread	Optional[bool]	If this is true, all other suspended threads are not resumed.	None
granularity	Optional[str]	The granularity of the step, assumed to be 'statement' if not specified.	None

Source code in src\dap\client.py

def next(
    self,
    thread_id: int,
    single_thread: Optional[bool] = None,
    granularity: Optional[str] = None,
) -> int:
    """The request steps through the program.

    Args:
        thread_id: Specifies the thread for which to resume execution for one step.
        single_thread: If this is true, all other suspended threads are not resumed.
        granularity: The granularity of the step, assumed to be 'statement' if not specified.
    """

    return self.send_request(
        "next",
        {
            "threadId": thread_id,
            "singleThread": single_thread,
            "granularity": granularity,
        },
    )

pause(thread_id)

Suspends the debuggee.

Parameters:

Name	Type	Description	Default
thread_id	int	The thread to pause.	required

Source code in src\dap\client.py

def pause(self, thread_id: int) -> int:
    """Suspends the debuggee.

    Args:
        thread_id: The thread to pause.
    """

    return self.send_request("pause", {"threadId": thread_id})

read_memory(memory_reference, count, offset=None)

Reads memory from the debuggee.

Parameters:

Name	Type	Description	Default
memory_reference	str	The memory reference to the base location from which to read memory.	required
count	int	The number of bytes to read at the specified location and offset.	required
offset	Optional[int]	The offset (in bytes) of the first byte to read.	None

Source code in src\dap\client.py

def read_memory(
    self, memory_reference: str, count: int, offset: Optional[int] = None
) -> int:
    """Reads memory from the debuggee.

    Args:
        memory_reference: The memory reference to the base location from which to read memory.
        count: The number of bytes to read at the specified location and offset.
        offset: The offset (in bytes) of the first byte to read.
    """

    return self.send_request(
        "readMemory",
        {"memoryReference": memory_reference, "offset": offset, "count": count},
    )

receive(data)

Feed data from the debug adapter to the client.

Parameters:

Name	Type	Description	Default
data	bytes	The data to receive.	required

Yields:

Type	Description
ResponseBody | EventBody	The response or event body.

Source code in src\dap\client.py

def receive(self, data: bytes) -> Generator[ResponseBody | EventBody, None, None]:
    """Feed data from the debug adapter to the client.

    Args:
        data: The data to receive.

    Yields:
        The response or event body.
    """

    self._receive_buf += data
    yield from self.handler.handle()

restart(arguments=None)

Restarts a debug session.

Parameters:

Name	Type	Description	Default
arguments	Optional[LaunchRequestArguments | AttachRequestArguments]	Use either arguments for the 'launch' or 'attach' request.	None

Source code in src\dap\client.py

def restart(
    self,
    arguments: Optional[LaunchRequestArguments | AttachRequestArguments] = None,
) -> int:
    """Restarts a debug session.

    Args:
        arguments: Use either arguments for the 'launch' or 'attach' request.
    """

    return self.send_request("restart", arguments)

restart_frame(frame_id)

Restart the stack frame identified by the given frame ID. The frame ID must have been obtained in the current suspended state.

Parameters:

Name	Type	Description	Default
frame_id	int	The frame to restart.	required

Source code in src\dap\client.py

def restart_frame(self, frame_id: int) -> int:
    """Restart the stack frame identified by the given frame ID.
    The frame ID must have been obtained in the current suspended state.

    Args:
        frame_id: The frame to restart.
    """

    return self.send_request("restartFrame", {"frameId": frame_id})

reverse_continue(thread_id, single_thread=None)

The request starts the debuggee to run backward.

Parameters:

Name	Type	Description	Default
thread_id	int	ID of the active thread.	required
single_thread	Optional[bool]	If true, backward execution is limited to the specified thread.	None

Source code in src\dap\client.py

def reverse_continue(
    self, thread_id: int, single_thread: Optional[bool] = None
) -> int:
    """The request starts the debuggee to run backward.

    Args:
        thread_id: ID of the active thread.
        single_thread: If true, backward execution is limited to the specified thread.
    """

    return self.send_request(
        "reverseContinue",
        {"threadId": thread_id, "singleThread": single_thread},
    )

scopes(frame_id)

The request returns the variable scopes for a given stack frame.

Parameters:

Name	Type	Description	Default
frame_id	int	Retrieve the scopes for this stackframe.	required

Source code in src\dap\client.py

def scopes(self, frame_id: int) -> int:
    """The request returns the variable scopes for a given stack frame.

    Args:
        frame_id: Retrieve the scopes for this stackframe.
    """

    return self.send_request("scopes", {"frameId": frame_id})

send()

Get the data to send to the debug adapter.

Returns:

Type	Description
bytes	The data to send.

Source code in src\dap\client.py

def send(self) -> bytes:
    """Get the data to send to the debug adapter.

    Returns:
        The data to send.
    """

    send_buf = self._send_buf
    self._send_buf = bytearray()
    return send_buf

send_request(command, arguments=None)

Send a request to the debug adapter.

This can be useful for sending requests that are not yet implemented in the client or for sending custom requests to the debug adapter that are specific to the adapter.

Parameters:

Name	Type	Description	Default
command	str	The command to send.	required
arguments	Optional[dict[str, Any]]	The arguments to send.	None

Returns:

Type	Description
int	The sequence number of the request.

Source code in src\dap\client.py

def send_request(
    self, command: str, arguments: Optional[dict[str, Any]] = None
) -> int:
    """Send a request to the debug adapter.

    This can be useful for sending requests that are not yet implemented in the client or
    for sending custom requests to the debug adapter that are specific to the adapter.

    Args:
        command: The command to send.
        arguments: The arguments to send.

    Returns:
        The sequence number of the request.
    """

    seq = self._seq
    self._seq += 1

    self._send_buf += RequestBuffer(seq, command, arguments)
    self._pending_requests[seq] = Request(
        seq=seq, command=command, arguments=arguments
    )
    return seq

set_breakpoints(source, breakpoints, lines=None, source_modified=None)

Sets multiple breakpoints for a single source and clears all previous breakpoints in that source.

Parameters:

Name	Type	Description	Default
source	Source	The source location of the breakpoints.	required
breakpoints	List[SourceBreakpoint]	The code locations of the breakpoints.	required
lines	Optional[List[int]]	Deprecated: The code locations of the breakpoints.	None
source_modified	Optional[bool]	A value of true indicates that the underlying source has been modified which results in new breakpoint locations.	None

Source code in src\dap\client.py

def set_breakpoints(
    self,
    source: Source,
    breakpoints: List[SourceBreakpoint],
    lines: Optional[List[int]] = None,
    source_modified: Optional[bool] = None,
) -> int:
    """Sets multiple breakpoints for a single source and clears all previous breakpoints in that source.

    Args:
        source: The source location of the breakpoints.
        breakpoints: The code locations of the breakpoints.
        lines: Deprecated: The code locations of the breakpoints.
        source_modified: A value of true indicates that the underlying source has been modified \
            which results in new breakpoint locations.
    """

    return self.send_request(
        "setBreakpoints",
        {
            "source": source,
            "breakpoints": breakpoints,
            "lines": lines,
            "sourceModified": source_modified,
        },
    )

set_data_breakpoints(breakpoints)

Replaces all existing data breakpoints with new data breakpoints.

Parameters:

Name	Type	Description	Default
breakpoints	List[DataBreakpoint]	The data breakpoints to set.	required

Source code in src\dap\client.py

def set_data_breakpoints(self, breakpoints: List[DataBreakpoint]) -> int:
    """Replaces all existing data breakpoints with new data breakpoints.

    Args:
        breakpoints: The data breakpoints to set.
    """

    return self.send_request("setDataBreakpoints", {"breakpoints": breakpoints})

set_exception_breakpoints(filters, filter_options, exception_options)

The request configures the debugger's response to thrown exceptions.

Each of the filters, filterOptions, and exceptionOptions in the request are independent configurations to a debug adapter indicating a kind of exception to catch. An exception thrown in a program should result in a stopped event from the debug adapter (with reason exception) if any of the configured filters match.

Parameters:

Name	Type	Description	Default
filters	List[str]	Set of exception filters specified by their ID.	required
filter_options	Optional[List[ExceptionFilterOptions]]	An array of ExceptionFilterOptions. The set of all possible exception filters is defined by the exceptionBreakpointFilters capability.	required
exception_options	Optional[List[ExceptionOptions]]	An array of ExceptionOptions. Configuration options for selected exceptions.	required

Source code in src\dap\client.py

def set_exception_breakpoints(
    self,
    filters: List[str],
    filter_options: Optional[List[ExceptionFilterOptions]],
    exception_options: Optional[List[ExceptionOptions]],
) -> int:
    """The request configures the debugger's response to thrown exceptions.

    Each of the filters, filterOptions, and exceptionOptions in the request are independent configurations
    to a debug adapter indicating a kind of exception to catch. An exception thrown in a program should result
    in a stopped event from the debug adapter (with reason exception) if any of the configured filters match.

    Args:
        filters: Set of exception filters specified by their ID.
        filter_options: An array of ExceptionFilterOptions. The set of all possible exception filters \
            is defined by the `exceptionBreakpointFilters` capability.
        exception_options: An array of ExceptionOptions. Configuration options for selected exceptions.
    """

    return self.send_request(
        "setExceptionBreakpoints",
        {
            "filters": filters,
            "filterOptions": filter_options,
            "exceptionOptions": exception_options,
        },
    )

set_expression(expression, value, frame_id=None, format=None)

Evaluates the given value expression and assigns it to the expression which must be a modifiable l-value.

The expressions have access to any variables and arguments that are in scope of the specified frame.

Parameters:

Name	Type	Description	Default
expression	str	The l-value expression to assign the result to.	required
value	str	The value expression to assign to the l-value expression.	required
frame_id	Optional[int]	Evaluate the expressions in the scope of this stack frame. If not specified, the expressions are evaluated in the global scope.	None
format	Optional[ValueFormat]	Specifies details on how to format the result.	None

Source code in src\dap\client.py

def set_expression(
    self,
    expression: str,
    value: str,
    frame_id: Optional[int] = None,
    format: Optional[ValueFormat] = None,
) -> int:
    """Evaluates the given value expression and assigns it to the expression which must be a modifiable l-value.

    The expressions have access to any variables and arguments that are in scope of the specified frame.

    Args:
        expression: The l-value expression to assign the result to.
        value: The value expression to assign to the l-value expression.
        frame_id: Evaluate the expressions in the scope of this stack frame. \
            If not specified, the expressions are evaluated in the global scope.
        format: Specifies details on how to format the result.
    """

    return self.send_request(
        "setExpression",
        {
            "expression": expression,
            "value": value,
            "frameId": frame_id,
            "format": format,
        },
    )

set_function_breakpoints(breakpoints=[])

Replaces all existing function breakpoints with new function breakpoints.

To clear all function breakpoints, call this without arguments. When a function breakpoint is hit, a stopped event (with reason function breakpoint) is generated.

Parameters:

Name	Type	Description	Default
breakpoints	List[FunctionBreakpoint]	The function breakpoints to set.	[]

Source code in src\dap\client.py

def set_function_breakpoints(
    self, breakpoints: List[FunctionBreakpoint] = []
) -> int:
    """Replaces all existing function breakpoints with new function breakpoints.

    To clear all function breakpoints, call this without arguments.
    When a function breakpoint is hit, a stopped event (with reason function breakpoint) is generated.

    Args:
        breakpoints: The function breakpoints to set.
    """

    return self.send_request("setFunctionBreakpoints", {"breakpoints": breakpoints})

set_instruction_breakpoints(breakpoints)

Replaces all existing instruction breakpoints. Typically, instruction breakpoints would be set from a disassembly window.

To clear all instruction breakpoints, specify an empty array. When an instruction breakpoint is hit, a stopped event (with reason instruction breakpoint) is generated.

Parameters:

Name	Type	Description	Default
breakpoints	List[InstructionBreakpoint]	The instruction breakpoints to set.	required

Source code in src\dap\client.py

def set_instruction_breakpoints(
    self, breakpoints: List[InstructionBreakpoint]
) -> int:
    """Replaces all existing instruction breakpoints. Typically, instruction breakpoints would be set from a disassembly window.

    To clear all instruction breakpoints, specify an empty array.
    When an instruction breakpoint is hit, a stopped event (with reason instruction breakpoint) is generated.

    Args:
        breakpoints: The instruction breakpoints to set.
    """

    return self.send_request(
        "setInstructionBreakpoints", {"breakpoints": breakpoints}
    )

set_variable(variables_reference, name, value, format=None)

Set the variable with the given name in the variable container to a new value.

Parameters:

Name	Type	Description	Default
variables_reference	int	The reference of the variable container.	required
name	str	The name of the variable to set.	required
value	str	The value to set.	required
format	Optional[ValueFormat]	Specifies details on how to format the response value.	None

Source code in src\dap\client.py

def set_variable(
    self,
    variables_reference: int,
    name: str,
    value: str,
    format: Optional[ValueFormat] = None,
) -> int:
    """Set the variable with the given name in the variable container to a new value.

    Args:
        variables_reference: The reference of the variable container.
        name: The name of the variable to set.
        value: The value to set.
        format: Specifies details on how to format the response value.
    """

    return self.send_request(
        "setVariable",
        {
            "variablesReference": variables_reference,
            "name": name,
            "value": value,
            "format": format,
        },
    )

source(source_reference, source=None)

The request retrieves the source code for a given source reference.

Parameters:

Name	Type	Description	Default
source_reference	int	The reference to the source. This is the same as source.sourceReference.	required
source	Optional[Source]	Specifies the source content to load. Either source.path or source.sourceReference must be specified.	None

Source code in src\dap\client.py

def source(self, source_reference: int, source: Optional[Source] = None) -> int:
    """The request retrieves the source code for a given source reference.

    Args:
        source_reference: The reference to the source. This is the same as `source.sourceReference`.
        source: Specifies the source content to load. Either `source.path` or `source.sourceReference` must be specified.
    """

    return self.send_request(
        "source", {"sourceReference": source_reference, "source": source}
    )

stack_trace(thread_id=None, start_frame=None, levels=None, format=None)

The request returns a stack trace from the current execution state.

Request all stack frames by omitting the startFrame and levels arguments.

Parameters:

Name	Type	Description	Default
thread_id	Optional[int]	Retrieve the stacktrace for this thread.	None
start_frame	Optional[int]	The index of the first frame to return; if omitted frames start at 0.	None
levels	Optional[int]	The maximum number of frames to return. If levels is not specified or 0, all frames are returned.	None
format	Optional[StackFrameFormat]	Specifies details on how to format the stack frames.	None

Source code in src\dap\client.py

def stack_trace(
    self,
    thread_id: Optional[int] = None,
    start_frame: Optional[int] = None,
    levels: Optional[int] = None,
    format: Optional[StackFrameFormat] = None,
) -> int:
    """The request returns a stack trace from the current execution state.

    Request all stack frames by omitting the startFrame and levels arguments.

    Args:
        thread_id: Retrieve the stacktrace for this thread.
        start_frame: The index of the first frame to return; if omitted frames start at 0.
        levels: The maximum number of frames to return. If levels is not specified or 0, all frames are returned.
        format: Specifies details on how to format the stack frames.
    """

    return self.send_request(
        "stackTrace",
        {
            "threadId": thread_id,
            "startFrame": start_frame,
            "levels": levels,
            "format": format,
        },
    )

step_back(thread_id, single_thread=None, granularity=None)

The request executes one backward step (in the given granularity) for the specified thread and allows all other threads to run backward freely by resuming them.

If the debug adapter supports single thread execution (see capability supportsSingleThreadExecutionRequests), setting the singleThread argument to true prevents other suspended threads from resuming. The debug adapter first sends the response and then a stopped event (with reason step) after the step has completed.

Parameters:

Name	Type	Description	Default
thread_id	int	ID of the active thread.	required
single_thread	Optional[bool]	If true, backward execution is limited to the specified thread.	None
granularity	Optional[SteppingGranularity]	The granularity of the step, assumed to be 'statement' if not specified.	None

Source code in src\dap\client.py

def step_back(
    self,
    thread_id: int,
    single_thread: Optional[bool] = None,
    granularity: Optional[SteppingGranularity] = None,
) -> int:
    """The request executes one backward step (in the given granularity) for the specified thread
    and allows all other threads to run backward freely by resuming them.

    If the debug adapter supports single thread execution (see capability `supportsSingleThreadExecutionRequests`),
    setting the singleThread argument to true prevents other suspended threads from resuming.
    The debug adapter first sends the response and then a stopped event (with reason step) after the step has completed.

    Args:
        thread_id: ID of the active thread.
        single_thread: If true, backward execution is limited to the specified thread.
        granularity: The granularity of the step, assumed to be 'statement' if not specified.
    """

    return self.send_request(
        "stepBack",
        {
            "threadId": thread_id,
            "singleThread": single_thread,
            "granularity": granularity,
        },
    )

step_in(thread_id, single_thread=None, target_id=None, granularity=None)

The request resumes the given thread to step into a function/method and allows all other threads to run freely by resuming them.

If the debug adapter supports single thread execution (see capability supportsSingleThreadExecutionRequests), setting the singleThread argument to true prevents other suspended threads from resuming.

If the request cannot step into a target, stepIn behaves like the next request. The debug adapter first sends the response and then a stopped event (with reason step) after the step has completed.

If there are multiple function/method calls (or other targets) on the source line, the argument targetId can be used to control into which target the stepIn should occur.

Parameters:

Name	Type	Description	Default
thread_id	int	ID of the active thread.	required
single_thread	Optional[bool]	If true, stepIn is limited to the specified thread.	None
target_id	Optional[int]	The stepIn target for this step.	None
granularity	Optional[SteppingGranularity]	The granularity of the step, assumed to be 'statement' if not specified.	None

Source code in src\dap\client.py

def step_in(
    self,
    thread_id: int,
    single_thread: Optional[bool] = None,
    target_id: Optional[int] = None,
    granularity: Optional[SteppingGranularity] = None,
) -> int:
    """The request resumes the given thread to step into a function/method and allows all other threads to run freely by resuming them.

    If the debug adapter supports single thread execution (see capability `supportsSingleThreadExecutionRequests`),
    setting the singleThread argument to true prevents other suspended threads from resuming.

    If the request cannot step into a target, stepIn behaves like the next request.
    The debug adapter first sends the response and then a stopped event (with reason step) after the step has completed.

    If there are multiple function/method calls (or other targets) on the source line,
    the argument targetId can be used to control into which target the stepIn should occur.

    Args:
        thread_id: ID of the active thread.
        single_thread: If true, stepIn is limited to the specified thread.
        target_id: The stepIn target for this step.
        granularity: The granularity of the step, assumed to be 'statement' if not specified.
    """

    return self.send_request(
        "stepIn",
        {
            "threadId": thread_id,
            "singleThread": single_thread,
            "targetId": target_id,
            "granularity": granularity,
        },
    )

step_in_targets(frame_id)

The request retrieves the possible stepIn targets for the specified stack frame. These targets can be used in the stepIn request.

Parameters:

Name	Type	Description	Default
frame_id	int	The stack frame for which to retrieve the possible stepIn targets.	required

Source code in src\dap\client.py

def step_in_targets(self, frame_id: int) -> int:
    """The request retrieves the possible stepIn targets for the specified stack frame.
    These targets can be used in the stepIn request.

    Args:
        frame_id: The stack frame for which to retrieve the possible stepIn targets.
    """

    return self.send_request("stepInTargets", {"frameId": frame_id})

step_out(thread_id, single_thread=None, granularity=None)

The request resumes the given thread to step out of the current function/method and allows all other threads to run freely by resuming them.

If the debug adapter supports single thread execution (see capability supportsSingleThreadExecutionRequests), setting the singleThread argument to true prevents other suspended threads from resuming.

The debug adapter first sends the response and then a stopped event (with reason step) after the step has completed.

Parameters:

Name	Type	Description	Default
thread_id	int	ID of the active thread.	required
single_thread	Optional[bool]	If true, stepOut is limited to the specified thread.	None
granularity	Optional[SteppingGranularity]	The granularity of the step, assumed to be 'statement' if not specified.	None

Source code in src\dap\client.py

def step_out(
    self,
    thread_id: int,
    single_thread: Optional[bool] = None,
    granularity: Optional[SteppingGranularity] = None,
) -> int:
    """The request resumes the given thread to step out of the current function/method and allows all other threads to run freely by resuming them.

    If the debug adapter supports single thread execution (see capability `supportsSingleThreadExecutionRequests`),
    setting the singleThread argument to true prevents other suspended threads from resuming.

    The debug adapter first sends the response and then a stopped event (with reason step) after the step has completed.

    Args:
        thread_id: ID of the active thread.
        single_thread: If true, stepOut is limited to the specified thread.
        granularity: The granularity of the step, assumed to be 'statement' if not specified.
    """

    return self.send_request(
        "stepOut",
        {
            "threadId": thread_id,
            "singleThread": single_thread,
            "granularity": granularity,
        },
    )

terminate(restart=None)

The terminate request is sent from the client to the debug adapter in order to shut down the debuggee gracefully.

Typically a debug adapter implements terminate by sending a software signal which the debuggee intercepts in order to clean things up properly before terminating itself.

Please note that this request does not directly affect the state of the debug session: if the debuggee decides to veto the graceful shutdown for any reason by not terminating itself, then the debug session just continues.

Parameters:

Name	Type	Description	Default
restart	Optional[bool]	A value of true indicates that this 'terminate' request is part of a restart sequence.	None

Source code in src\dap\client.py

def terminate(self, restart: Optional[bool] = None) -> int:
    """The terminate request is sent from the client to the debug adapter in order to shut down the debuggee gracefully.

    Typically a debug adapter implements terminate by sending a software signal which the debuggee intercepts in order
    to clean things up properly before terminating itself.

    Please note that this request does not directly affect the state of the debug session: if the debuggee decides to
    veto the graceful shutdown for any reason by not terminating itself, then the debug session just continues.

    Args:
        restart: A value of true indicates that this 'terminate' request is part of a restart sequence.
    """

    return self.send_request("terminate", {"restart": restart})

terminate_threads(thread_ids)

The request terminates the threads with the given ids.

Parameters:

Name	Type	Description	Default
thread_ids	List[int]	The threads to terminate.	required

Source code in src\dap\client.py

def terminate_threads(self, thread_ids: List[int]) -> int:
    """The request terminates the threads with the given ids.

    Args:
        thread_ids: The threads to terminate.
    """

    return self.send_request("terminateThreads", {"threadIds": thread_ids})

threads()

The request retrieves a list of all threads.

Source code in src\dap\client.py

def threads(self) -> int:
    """The request retrieves a list of all threads."""

    return self.send_request("threads")

variables(variables_reference, filter=None, start=None, count=None, format=None)

Retrieves all child variables for the given variable reference.

A filter can be used to limit the fetched children to either named or indexed children.

Parameters:

Name	Type	Description	Default
variables_reference	int	The variable for which to retrieve its children.	required
filter	Optional[Literal[indexed, named]] | str	Filter to limit the child variables to either named or indexed. If not specified, both types are fetched.	None
start	Optional[int]	The index of the first variable to return; if omitted variables start at 0.	None
count	Optional[int]	The number of variables to return. If not passed or 0, all variables are returned.	None
format	Optional[ValueFormat]	Specifies details on how to format the response value.	None

Source code in src\dap\client.py

def variables(
    self,
    variables_reference: int,
    filter: Optional[Literal["indexed", "named"]] | str = None,
    start: Optional[int] = None,
    count: Optional[int] = None,
    format: Optional[ValueFormat] = None,
) -> int:
    """Retrieves all child variables for the given variable reference.

    A filter can be used to limit the fetched children to either named or indexed children.

    Args:
        variables_reference: The variable for which to retrieve its children.
        filter: Filter to limit the child variables to either named or indexed. If not specified, both types are fetched.
        start: The index of the first variable to return; if omitted variables start at 0.
        count: The number of variables to return. If not passed or 0, all variables are returned.
        format: Specifies details on how to format the response value.
    """

    return self.send_request(
        "variables",
        {
            "variablesReference": variables_reference,
            "filter": filter,
            "start": start,
            "count": count,
            "format": format,
        },
    )

write_memory(memory_reference, data, offset=None, allow_partial=None)

Writes bytes to memory at the provided location.

Parameters:

Name	Type	Description	Default
memory_reference	str	The memory reference to the base location to write memory.	required
data	str	Bytes to write, encoded using base64.	required
offset	Optional[int]	The offset (in bytes) of the first byte to write. Can be negative.	None
allow_partial	Optional[bool]	Property to control partial writes. If true, the debug adapter should attempt to write memory even if the entire memory region is not writable.	None

Source code in src\dap\client.py

def write_memory(
    self,
    memory_reference: str,
    data: str,
    offset: Optional[int] = None,
    allow_partial: Optional[bool] = None,
) -> int:
    """Writes bytes to memory at the provided location.

    Args:
        memory_reference: The memory reference to the base location to write memory.
        data: Bytes to write, encoded using base64.
        offset: The offset (in bytes) of the first byte to write. Can be negative.
        allow_partial: Property to control partial writes. If true, the debug adapter should \
            attempt to write memory even if the entire memory region is not writable.
    """

    return self.send_request(
        "writeMemory",
        {
            "memoryReference": memory_reference,
            "offset": offset,
            "data": data,
            "allowPartial": allow_partial,
        },
    )

Responses

Attached

Bases: Response

Response to 'attach' request.

Source code in src\dap\responses.py

class Attached(Response):
    """Response to 'attach' request."""

    ...

BreakpointLocationsResponse

Bases: ResponseBody

Body of a 'breakpointLocations' response.

Source code in src\dap\responses.py

class BreakpointLocationsResponse(ResponseBody):
    """Body of a 'breakpointLocations' response."""

    breakpoints: list[BreakpointLocation] = Field(
        ..., description="List of breakpoints."
    )

Cancelled

Bases: Response

Response to 'cancel' request.

Source code in src\dap\responses.py

class Cancelled(Response):
    """Response to 'cancel' request."""

    ...

CompletionsResponse

Bases: ResponseBody

Body of a 'completions' response.

Source code in src\dap\responses.py

class CompletionsResponse(ResponseBody):
    """Body of a 'completions' response."""

    targets: list[CompletionItem] = Field(..., description="List of completion items.")

ConfigurationDone

Bases: Response

Response to 'configurationDone' request.

Source code in src\dap\responses.py

class ConfigurationDone(Response):
    """Response to 'configurationDone' request."""

    ...

Continued

Bases: ResponseBody

Body of a 'continue' response.

Source code in src\dap\responses.py

class Continued(ResponseBody):
    """Body of a 'continue' response."""

    allThreadsContinued: Optional[bool] = Field(
        None, description="If all threads were continued."
    )

DataBreakpointInfoResponse

Bases: ResponseBody

Body of a 'dataBreakpoint' response.

Source code in src\dap\responses.py

class DataBreakpointInfoResponse(ResponseBody):
    """Body of a 'dataBreakpoint' response."""

    dataId: Optional[str] = Field(
        None, description="An identifier for the data breakpoint."
    )
    description: str = Field(
        ..., description="A user-visible description of the breakpoint."
    )
    accessTypes: Optional[list[DataBreakpointAccessType]] = Field(
        None, description="The access types of the data breakpoint."
    )
    canPersist: Optional[bool] = Field(
        None, description="Whether the data breakpoint can be persisted."
    )

DisassembleResponse

Bases: ResponseBody

Body of a 'disassemble' response.

Source code in src\dap\responses.py

class DisassembleResponse(ResponseBody):
    """Body of a 'disassemble' response."""

    instructions: list[DisassembledInstruction] = Field(
        ..., description="List of disassembled instructions."
    )

Disconnected

Bases: Response

Response to 'disconnect' request.

Source code in src\dap\responses.py

class Disconnected(Response):
    """Response to 'disconnect' request."""

    ...

EvaluateResponse

Bases: ResponseBody

Body of an 'evaluate' response.

Source code in src\dap\responses.py

class EvaluateResponse(ResponseBody):
    """Body of an 'evaluate' response."""

    result: str = Field(..., description="The result of the evaluation.")
    type: Optional[str] = Field(None, description="The type of the result.")
    presentationHint: Optional[VariablePresentationHint] = Field(
        None, description="The presentation hint of the result."
    )
    variablesReference: int = Field(
        ...,
        description="The reference to the variables of the evaluation result.",
    )
    namedVariables: Optional[int] = Field(
        None, description="The number of named variables."
    )
    indexedVariables: Optional[int] = Field(
        None, description="The number of indexed variables."
    )
    memoryReference: Optional[str] = Field(
        None, description="The memory reference of the result."
    )

GotoDone

Bases: Response

Response to 'goto' request.

Source code in src\dap\responses.py

class GotoDone(Response):
    """Response to 'goto' request."""

    ...

GotoTargetsResponse

Bases: ResponseBody

Body of a 'gotoTargets' response.

Source code in src\dap\responses.py

class GotoTargetsResponse(ResponseBody):
    """Body of a 'gotoTargets' response."""

    targets: list[GotoTarget] = Field(..., description="List of goto targets.")

Initialized

Bases: Capabilities

Response to 'initialize' request.

Source code in src\dap\responses.py

class Initialized(Capabilities):
    """Response to 'initialize' request."""

    ...

LaunchDone

Bases: Response

Response to 'launch' request.

Source code in src\dap\responses.py

class LaunchDone(Response):
    """Response to 'launch' request."""

    ...

LoadedSourcesResponse

Bases: ResponseBody

Body of a 'loadedSources' response.

Source code in src\dap\responses.py

class LoadedSourcesResponse(ResponseBody):
    """Body of a 'loadedSources' response."""

    sources: list[Source] = Field(..., description="List of loaded sources.")

ModulesResponse

Bases: ResponseBody

Body of a 'modules' response.

Source code in src\dap\responses.py

class ModulesResponse(ResponseBody):
    """Body of a 'modules' response."""

    modules: list[Module] = Field(..., description="List of modules.")
    totalModules: Optional[int] = Field(
        None, description="The total number of modules."
    )

NextResponse

Bases: Response

Response to 'next' request.

Source code in src\dap\responses.py

class NextResponse(Response):
    """Response to 'next' request."""

    ...

Paused

Bases: Response

Response to 'pause' request.

Source code in src\dap\responses.py

class Paused(Response):
    """Response to 'pause' request."""

    ...

ReadMemoryResponse

Bases: ResponseBody

Body of a 'readMemory' response.

Source code in src\dap\responses.py

class ReadMemoryResponse(ResponseBody):
    """Body of a 'readMemory' response."""

    address: str = Field(..., description="The address of the memory read.")
    unreadableBytes: Optional[int] = Field(
        None, description="The number of unreadable bytes."
    )
    data: Optional[str] = Field(None, description="The data read from memory.")

RestartFrameDone

Bases: Response

Response to 'restartFrame' request.

Source code in src\dap\responses.py

class RestartFrameDone(Response):
    """Response to 'restartFrame' request."""

    ...

Restarted

Bases: Response

Response to 'restart' request.

Source code in src\dap\responses.py

class Restarted(Response):
    """Response to 'restart' request."""

    ...

ReverseContinueDone

Bases: Response

Response to 'reverseContinue' request.

Source code in src\dap\responses.py

class ReverseContinueDone(Response):
    """Response to 'reverseContinue' request."""

    ...

RunInTerminalResponse

Bases: Response

Response to 'runInTerminal' request.

Source code in src\dap\responses.py

class RunInTerminalResponse(Response):
    """Response to 'runInTerminal' request."""

    request_seq: int = Field(
        ..., description="Sequence number of the corresponding request."
    )
    success: bool = Field(
        ..., description="Indicates whether the request was successful."
    )
    command: str = "runInTerminal"
    body: RunInTerminalResponseBody

RunInTerminalResponseBody

Bases: ResponseBody

Body of a 'runInTerminal' response.

Source code in src\dap\responses.py

class RunInTerminalResponseBody(ResponseBody):
    """Body of a 'runInTerminal' response."""

    processId: Optional[int] = Field(
        None, description="The process ID of the terminal."
    )
    shellProcessId: Optional[int] = Field(
        None, description="The process ID of the shell."
    )

ScopesResponse

Bases: ResponseBody

Body of a 'scopes' response.

Source code in src\dap\responses.py

class ScopesResponse(ResponseBody):
    """Body of a 'scopes' response."""

    scopes: list[Scope] = Field(..., description="List of scopes.")

SetBreakpointsResponse

Bases: ResponseBody

Body of a 'setBreakpoints' response.

Source code in src\dap\responses.py

class SetBreakpointsResponse(ResponseBody):
    """Body of a 'setBreakpoints' response."""

    breakpoints: list[Breakpoint]

SetDataBreakpointsResponse

Bases: ResponseBody

Body of a 'setDataBreakpoints' response.

Source code in src\dap\responses.py

class SetDataBreakpointsResponse(ResponseBody):
    """Body of a 'setDataBreakpoints' response."""

    breakpoints: list[Breakpoint]

SetExceptionBreakpointsResponse

Bases: Response

Response to 'setExceptionBreakpoints' request.

Source code in src\dap\responses.py

class SetExceptionBreakpointsResponse(Response):
    """Response to 'setExceptionBreakpoints' request."""

    ...

SetExpressionResponse

Bases: ResponseBody

Body of a 'setExpression' response.

Source code in src\dap\responses.py

class SetExpressionResponse(ResponseBody):
    """Body of a 'setExpression' response."""

    value: str = Field(..., description="The value of the expression.")
    type: Optional[str] = Field(None, description="The type of the expression.")
    presentationHint: Optional[VariablePresentationHint] = Field(
        None, description="The presentation hint of the expression."
    )
    variablesReference: Optional[int] = Field(
        None,
        description="The reference to the variables of the expression result.",
    )
    namedVariables: Optional[int] = Field(
        None, description="The number of named variables."
    )
    indexedVariables: Optional[int] = Field(
        None, description="The number of indexed variables."
    )
    memoryReference: Optional[str] = Field(
        None, description="The memory reference of the expression."
    )

SetFunctionBreakpointsResponse

Bases: ResponseBody

Body of a 'setFunctionBreakpoints' response.

Source code in src\dap\responses.py

class SetFunctionBreakpointsResponse(ResponseBody):
    """Body of a 'setFunctionBreakpoints' response."""

    breakpoints: list[FunctionBreakpoint] = Field(
        ..., description="List of function breakpoints."
    )

SetInstructionBreakpointsResponse

Bases: ResponseBody

Body of a 'setInstructionBreakpoints' response.

Source code in src\dap\responses.py

class SetInstructionBreakpointsResponse(ResponseBody):
    """Body of a 'setInstructionBreakpoints' response."""

    breakpoints: list[Breakpoint]

SetVariableResponse

Bases: ResponseBody

Body of a 'setVariable' response.

Source code in src\dap\responses.py

class SetVariableResponse(ResponseBody):
    """Body of a 'setVariable' response."""

    value: str = Field(..., description="The value of the variable.")
    type: Optional[str] = Field(None, description="The type of the variable.")
    variablesReference: Optional[int] = Field(
        None,
        description="The reference to the variables of the variable result.",
    )
    namedVariables: Optional[int] = Field(
        None, description="The number of named variables."
    )
    indexedVariables: Optional[int] = Field(
        None, description="The number of indexed variables."
    )
    memoryReference: Optional[str] = Field(
        None, description="The memory reference of the variable."
    )

SourceResponse

Bases: ResponseBody

Body of a 'source' response.

Source code in src\dap\responses.py

class SourceResponse(ResponseBody):
    """Body of a 'source' response."""

    content: str = Field(..., description="The content of the source.")
    mimeType: Optional[str] = Field(None, description="The MIME type of the source.")

StackTraceResponse

Bases: ResponseBody

Body of a 'stackTrace' response.

Source code in src\dap\responses.py

class StackTraceResponse(ResponseBody):
    """Body of a 'stackTrace' response."""

    stackFrames: list[StackFrame] = Field(..., description="List of stack frames.")
    totalFrames: Optional[int] = Field(None, description="The total number of frames.")

StartDebuggingResponse

Bases: Response

Body of a 'startDebugging' response.

Source code in src\dap\responses.py

class StartDebuggingResponse(Response):
    """Body of a 'startDebugging' response."""

    request_seq: int = Field(
        ..., description="Sequence number of the corresponding request."
    )
    success: bool = Field(
        ..., description="Indicates whether the request was successful."
    )
    command: str = "startDebugging"

StepBackDone

Bases: Response

Response to 'stepBack' request.

Source code in src\dap\responses.py

class StepBackDone(Response):
    """Response to 'stepBack' request."""

    ...

StepInDone

Bases: Response

Response to 'stepIn' request.

Source code in src\dap\responses.py

class StepInDone(Response):
    """Response to 'stepIn' request."""

    ...

StepInTargetsResponse

Bases: ResponseBody

Body of a 'stepInTargets' response.

Source code in src\dap\responses.py

class StepInTargetsResponse(ResponseBody):
    """Body of a 'stepInTargets' response."""

    targets: list[StepInTarget] = Field(..., description="List of step in targets.")

StepOutDone

Bases: Response

Response to 'stepOut' request.

Source code in src\dap\responses.py

class StepOutDone(Response):
    """Response to 'stepOut' request."""

    ...

TerminateThreadsDone

Bases: Response

Response to 'terminateThreads' request.

Source code in src\dap\responses.py

class TerminateThreadsDone(Response):
    """Response to 'terminateThreads' request."""

    ...

Terminated

Bases: Response

Response to 'terminate' request.

Source code in src\dap\responses.py

class Terminated(Response):
    """Response to 'terminate' request."""

    ...

ThreadsResponse

Bases: ResponseBody

Body of a 'threads' response.

Source code in src\dap\responses.py

class ThreadsResponse(ResponseBody):
    """Body of a 'threads' response."""

    threads: list[Thread] = Field(..., description="List of threads.")

VariablesResponse

Bases: ResponseBody

Body of a 'variables' response.

Source code in src\dap\responses.py

class VariablesResponse(ResponseBody):
    """Body of a 'variables' response."""

    variables: list[Variable] = Field(..., description="List of variables.")

WriteMemoryResponse

Bases: ResponseBody

Body of a 'writeMemory' response.

Source code in src\dap\responses.py

class WriteMemoryResponse(ResponseBody):
    """Body of a 'writeMemory' response."""

    offset: Optional[int] = Field(None, description="The offset of the memory write.")
    bytesWritten: Optional[int] = Field(
        None, description="The number of bytes written."
    )

Events

BreakpointEvent

Bases: EventBody

Event sent when a breakpoint is hit.

Source code in src\dap\events.py

class BreakpointEvent(EventBody):
    """Event sent when a breakpoint is hit."""

    reason: Literal["changed", "new", "removed"] | str = Field(
        ...,
        description="The reason the event was sent. 'changed' for a changed breakpoint, 'new' for a new breakpoint, 'removed' for a removed breakpoint.",
    )
    breakpoint: Breakpoint = Field(..., description="The breakpoint that was hit.")

CapabilitiesEvent

Bases: EventBody

Event sent when capabilities are requested.

Source code in src\dap\events.py

class CapabilitiesEvent(EventBody):
    """Event sent when capabilities are requested."""

    capabilities: Capabilities = Field(
        ..., description="The capabilities of the debug adapter."
    )

ContinuedEvent

Bases: EventBody

Event sent when the execution is continued.

Source code in src\dap\events.py

class ContinuedEvent(EventBody):
    """Event sent when the execution is continued."""

    threadId: int = Field(..., description="The thread that continued.")
    allThreadsContinued: Optional[bool] = Field(
        None,
        description="If all threads continued, the attribute is omitted. If a specific thread continued, the attribute contains the thread ID.",
    )

ExitedEvent

Bases: EventBody

Event sent when the debuggee has exited.

Source code in src\dap\events.py

class ExitedEvent(EventBody):
    """Event sent when the debuggee has exited."""

    exitCode: int = Field(..., description="The exit code of the debuggee.")

InitializedEvent

Bases: EventBody

Event sent when the debug adapter is initialized.

Source code in src\dap\events.py

class InitializedEvent(EventBody):
    """Event sent when the debug adapter is initialized."""

    ...

InvalidatedEvent

Bases: EventBody

Event sent when breakpoints are invalidated.

Source code in src\dap\events.py

class InvalidatedEvent(EventBody):
    """Event sent when breakpoints are invalidated."""

    areas: Optional[list[InvalidatedAreas]] = Field(
        None, description="The invalidated areas."
    )
    threadId: Optional[int] = Field(
        None,
        description="The thread ID of the invalidated areas. If omitted, the invalidated areas are not specific to a thread.",
    )
    stackFrameId: Optional[int] = Field(
        None,
        description="The stack frame ID of the invalidated areas. If omitted, the invalidated areas are not specific to a stack frame.",
    )

LoadedSourceEvent

Bases: EventBody

Event sent when a source is loaded.

Source code in src\dap\events.py

class LoadedSourceEvent(EventBody):
    """Event sent when a source is loaded."""

    reason: Literal["new", "changed", "removed"] = Field(
        ...,
        description="The reason the event was sent. 'new' for a new source, 'changed' for a changed source, 'removed' for a removed source.",
    )
    source: Source = Field(..., description="The source that was loaded.")

MemoryEvent

Bases: EventBody

Event sent when memory is accessed.

Source code in src\dap\events.py

class MemoryEvent(EventBody):
    """Event sent when memory is accessed."""

    memoryReference: str = Field(..., description="The memory reference.")
    offset: int = Field(..., description="The memory offset.")
    count: int = Field(..., description="The number of bytes.")

ModuleEvent

Bases: EventBody

Event sent when a module is loaded or unloaded.

Source code in src\dap\events.py

class ModuleEvent(EventBody):
    """Event sent when a module is loaded or unloaded."""

    reason: Literal["new", "changed", "removed"] = Field(
        ...,
        description="The reason the event was sent. 'new' for a new module, 'changed' for a changed module, 'removed' for a removed module.",
    )
    module: Module = Field(..., description="The module that was loaded or unloaded.")

OutputEvent

Bases: EventBody

Event sent when output is produced.

Source code in src\dap\events.py

class OutputEvent(EventBody):
    """Event sent when output is produced."""

    category: Optional[
        Literal["console", "important", "stdout", "stderr", "telemetry"] | str
    ] = Field(
        ...,
        description="The category of the output.",
    )
    output: str = Field(..., description="The output text.")
    group: Optional[Literal["start", "startCollapsed", "end"]] = Field(
        None,
        description="The output group.",
    )
    variablesReference: Optional[int] = Field(
        None,
        description="The reference to a variable.",
    )
    source: Optional[Source] = Field(
        None,
        description="The source of the output.",
    )
    line: Optional[int] = Field(
        None,
        description="The line of the output.",
    )
    column: Optional[int] = Field(
        None,
        description="The column of the output.",
    )
    data: Optional[Any] = Field(None, description="Additional data.")

ProcessEvent

Bases: EventBody

Event sent when a process is created or exited.

Source code in src\dap\events.py

class ProcessEvent(EventBody):
    """Event sent when a process is created or exited."""

    name: str = Field(..., description="The name of the process.")
    systemProcessId: Optional[int] = Field(
        None,
        description="The system process ID.",
    )
    isLocalProcess: Optional[bool] = Field(
        None,
        description="If the process is local.",
    )
    startMethod: Optional[Literal["launch", "attach", "attachForSuspendedLaunch"]] = (
        Field(
            None,
            description="The start method.",
        )
    )
    pointerSize: Optional[int] = Field(
        None,
        description="The pointer size.",
    )

ProgressEndEvent

Bases: EventBody

Event sent when a progress ends.

Source code in src\dap\events.py

class ProgressEndEvent(EventBody):
    """Event sent when a progress ends."""

    progressId: str = Field(..., description="The progress ID.")
    message: Optional[str] = Field(None, description="The message.")

ProgressStartEvent

Bases: EventBody

Event sent when a progress starts.

Source code in src\dap\events.py

class ProgressStartEvent(EventBody):
    """Event sent when a progress starts."""

    progressId: str = Field(..., description="The progress ID.")
    title: str = Field(..., description="The title of the progress.")
    requestId: Optional[str] = Field(
        None,
        description="The request ID.",
    )
    cancellable: Optional[bool] = Field(
        None,
        description="If the progress is cancellable.",
    )
    message: Optional[str] = Field(None, description="The message.")
    percentage: Optional[int] = Field(
        None,
        description="The percentage of the progress.",
    )

ProgressUpdateEvent

Bases: EventBody

Event sent when a progress updates.

Source code in src\dap\events.py

class ProgressUpdateEvent(EventBody):
    """Event sent when a progress updates."""

    progressId: str = Field(..., description="The progress ID.")
    message: Optional[str] = Field(None, description="The message.")
    percentage: Optional[int] = Field(
        None,
        description="The percentage of the progress.",
    )

StoppedEvent

Bases: EventBody

Event sent when the execution is stopped.

Source code in src\dap\events.py

class StoppedEvent(EventBody):
    """Event sent when the execution is stopped."""

    reason: Literal[
        "step",
        "breakpoint",
        "exception",
        "pause",
        "entry",
        "goto",
        "function breakpoint",
        "data breakpoint",
        "instruction breakpoint",
    ] = Field(
        ...,
        description="The reason the execution was stopped.",
    )
    description: Optional[str] = Field(
        None,
        description="The description of the stop event.",
    )
    threadId: Optional[int] = Field(
        None,
        description="The thread that stopped.",
    )
    preserveFocusHint: Optional[bool] = Field(
        None,
        description="If the focus should be preserved.",
    )
    text: Optional[str] = Field(
        None,
        description="The text of the stop event.",
    )
    allThreadsStopped: Optional[bool] = Field(
        None,
        description="If all threads stopped, the attribute is omitted. If a specific thread stopped, the attribute contains the thread ID.",
    )
    hitBreakpointIds: Optional[list[str]] = Field(
        None,
        description="The hit breakpoint IDs.",
    )

TerminatedEvent

Bases: EventBody

Event sent when the debuggee is terminated.

Source code in src\dap\events.py

class TerminatedEvent(EventBody):
    """Event sent when the debuggee is terminated."""

    restart: Optional[bool] = Field(
        None,
        description="If the debuggee should be restarted.",
    )

ThreadEvent

Bases: EventBody

Event sent when a thread is created or exited.

Source code in src\dap\events.py

class ThreadEvent(EventBody):
    """Event sent when a thread is created or exited."""

    reason: Literal["started", "exited"] | str = Field(
        ...,
        description="The reason the event was sent. 'started' for a started thread, 'exited' for an exited thread.",
    )
    threadId: int = Field(..., description="The thread ID.")

Requests

AttachRequestArguments

Bases: TypedDict

Arguments for 'attach' request.

Source code in src\dap\requests.py

class AttachRequestArguments(TypedDict):
    """Arguments for 'attach' request."""

    __restart: Optional[bool] = None

LaunchRequestArguments

Bases: TypedDict

Arguments for 'launch' request.

Source code in src\dap\requests.py

class LaunchRequestArguments(TypedDict):
    """Arguments for 'launch' request."""

    noDebug: Optional[bool] = None
    __restart: Optional[bool] = None

RunInTerminalRequest

Bases: Request

Request for 'runInTerminal' reverse request.

Source code in src\dap\requests.py

class RunInTerminalRequest(Request):
    """Request for 'runInTerminal' reverse request."""

    command: Literal["runInTerminal"] = "runInTerminal"
    arguments: RunInTerminalRequestArguments

    def reply(
        self,
        success: bool,
        processId: Optional[int] = None,
        shellProcessId: Optional[int] = None,
    ) -> RunInTerminalResponse:
        return RunInTerminalResponse(
            seq=self.seq,
            request_seq=self.seq,
            success=success,
            body=RunInTerminalResponseBody(
                processId=processId,
                shellProcessId=shellProcessId,
            ),
        )

RunInTerminalRequestArguments

Bases: RequestArguments

Arguments for 'runInTerminal' reverse request.

Source code in src\dap\requests.py

class RunInTerminalRequestArguments(RequestArguments):
    """Arguments for 'runInTerminal' reverse request."""

    kind: Optional[str] = Field(
        None,
        description="The kind of terminal to run the command in.",
    )
    title: Optional[str] = Field(
        None,
        description="The title of the terminal.",
    )
    cwd: str = Field(
        ...,
        description="The working directory of the terminal.",
    )
    args: list[str] = Field(
        ...,
        description="The command to run in the terminal.",
    )
    env: Optional[dict[str, str | None]] = Field(
        None,
        description="Environment key-value pairs that should be added to or removed from the default environment.",
    )
    argsCanBeInterpretedByShell: Optional[bool] = Field(
        None,
        description="Whether the arguments can be interpreted as shell commands.",
    )

StartDebuggingRequest

Bases: Request

Request for 'startDebugging' request.

Source code in src\dap\requests.py

class StartDebuggingRequest(Request):
    """Request for 'startDebugging' request."""

    command: Literal["startDebugging"] = "startDebugging"
    arguments: StartDebuggingRequestArguments

    def reply(self, success: bool) -> StartDebuggingResponse:
        return StartDebuggingResponse(
            seq=self.seq,
            request_seq=self.seq,
            success=success,
        )

StartDebuggingRequestArguments

Bases: RequestArguments

Arguments for 'startDebugging' request.

Source code in src\dap\requests.py

class StartDebuggingRequestArguments(RequestArguments):
    """Arguments for 'startDebugging' request."""

    configuration: dict[str, str] = Field(
        ...,
        description="Arguments passed to the new debug session. The arguments must only contain properties understood by "
        "the `launch` or `attach` requests of the debug adapter and they must not contain any client-specific properties "
        "(e.g. `type`) or client-specific features (e.g. substitutable 'variables')",
    )
    request: Literal["launch", "attach"] = Field(
        ...,
        description="The request type: 'launch' or 'attach'.",
    )

Base Protocol

EventBody

Bases: BaseModel

Base class of event bodies

Source code in src\dap\base.py

class EventBody(BaseModel):
    """Base class of event bodies"""

    ...

Events

Bases: StrEnum

Enumeration of DAP events.

Source code in src\dap\base.py

class Events(StrEnum):
    """Enumeration of DAP events."""

    BREAKPOINT = "breakpoint"
    CAPABILITIES = "capabilities"
    CONTINUED = "continued"
    EXITED = "exited"
    INITIALIZED = "initialized"
    INVALIDATED = "invalidated"
    LOADED_SOURCE = "loadedSource"
    MEMORY = "memory"
    MODULE = "module"
    OUTPUT = "output"
    PROCESS = "process"
    PROGRESS_END = "progressEnd"
    PROGRESS_START = "progressStart"
    PROGRESS_UPDATE = "progressUpdate"
    STOPPED = "stopped"
    TERMINATED = "terminated"
    THREAD = "thread"

ProtocolMessage

Bases: BaseModel

Base class of requests, responses, and events

Source code in src\dap\base.py

class ProtocolMessage(BaseModel):
    """Base class of requests, responses, and events"""

    seq: int = Field(..., description="Sequence number (message ID) of the message.")
    type: Literal["request", "response", "event"] | str = Field(
        ..., description="Message type."
    )

Request

Bases: ProtocolMessage

Source code in src\dap\base.py

class Request(ProtocolMessage):
    type: str = "request"  # type: ignore
    command: str = Field(..., description="The command to execute.")
    arguments: Optional[Any] = Field(
        None, description="Object containing arguments for the command."
    )

    def reply(self, *a, **kw) -> Response:
        """Create a response message for the request.

        This method is for reverse requests only."""
        ...

reply(*a, **kw)

Create a response message for the request.

This method is for reverse requests only.

Source code in src\dap\base.py

def reply(self, *a, **kw) -> Response:
    """Create a response message for the request.

    This method is for reverse requests only."""
    ...

RequestArguments

Bases: BaseModel

Base class of request arguments

Source code in src\dap\base.py

class RequestArguments(BaseModel):
    """Base class of request arguments"""

Requests

Bases: StrEnum

Enumeration of DAP requests.

Source code in src\dap\base.py

class Requests(StrEnum):
    """Enumeration of DAP requests."""

    CANCEL = "cancel"
    ATTACH = "attach"
    BREAKPOINTLOCATIONS = "breakpointLocations"
    COMPLETIONS = "completions"
    CONFIGURATIONDONE = "configurationDone"
    CONTINUE = "continue"
    DATABREAKPOINTINFO = "dataBreakpointInfo"
    DISASSEMBLE = "disassemble"
    DISCONNECT = "disconnect"
    EVALUATE = "evaluate"
    EXCEPTIONINFO = "exceptionInfo"
    GOTO = "goto"
    GOTOTARGETS = "gotoTargets"
    INITIALIZE = "initialize"
    LAUNCH = "launch"
    LOADEDSOURCES = "loadedSources"
    MODULES = "modules"
    NEXT = "next"
    PAUSE = "pause"
    READMEMORY = "readMemory"
    RESTART = "restart"
    RESTARTFRAME = "restartFrame"
    REVERSECONTINUE = "reverseContinue"
    SCOPES = "scopes"
    SETBREAKPOINTS = "setBreakpoints"
    SETDATABREAKPOINTS = "setDataBreakpoints"
    SETEXCEPTIONBREAKPOINTS = "setExceptionBreakpoints"
    SETEXPRESSION = "setExpression"
    SETFUNCTIONBREAKPOINTS = "setFunctionBreakpoints"
    SETINSTRUCTIONBREAKPOINTS = "setInstructionBreakpoints"
    SETVARIABLE = "setVariable"
    SOURCE = "source"
    STACKTRACE = "stackTrace"
    STEPBACK = "stepBack"
    STEPIN = "stepIn"
    STEPINTARGETS = "stepInTargets"
    STEPOUT = "stepOut"
    TERMINATE = "terminate"
    TERMINATETHREADS = "terminateThreads"
    THREADS = "threads"
    VARIABLES = "variables"
    WRITEMEMORY = "writeMemory"

    # Reverse requests
    RUNINTERMINAL = "runInTerminal"
    STARTDEBUGGING = "startDebugging"

ResponseBody

Bases: BaseModel

Base class of response bodies

Source code in src\dap\base.py

class ResponseBody(BaseModel):
    """Base class of response bodies"""

    ...

Connection

AsyncConnection

Asyncio-based connection to a debug adapter server.

This class is used to connect to a debug adapter server using asyncio. It provides methods to start, stop, read and write to the server.

Source code in src\dap\connection.py

class AsyncConnection:
    """Asyncio-based connection to a debug adapter server.

    This class is used to connect to a debug adapter server using asyncio.
    It provides methods to start, stop, read and write to the server."""

    def __init__(self, host: str, port: int) -> None:
        self.host = host
        self.port = port
        self.reader: Optional[asyncio.StreamReader] = None
        self.writer: Optional[asyncio.StreamWriter] = None
        self.alive = False

    async def start(self):
        """Start the connection to the server."""

        self.reader, self.writer = await asyncio.open_connection(self.host, self.port)
        self.alive = True

    async def stop(self):
        """Stop the connection to the server."""

        self.writer.close()
        await self.writer.wait_closed()
        self.alive = False

    async def write(self, data: bytes):
        """Write data to the server

        Args:
            data (bytes): The data to write to the server.
        """

        self.writer.write(data)
        await self.writer.drain()

    async def read(self) -> bytes:
        """Read data from the server

        Returns:
            bytes: The data read from the server.
        """

        return await self.reader.read(1024)

read() async

Read data from the server

Returns:

Name	Type	Description
bytes	bytes	The data read from the server.

Source code in src\dap\connection.py

async def read(self) -> bytes:
    """Read data from the server

    Returns:
        bytes: The data read from the server.
    """

    return await self.reader.read(1024)

start() async

Start the connection to the server.

Source code in src\dap\connection.py

async def start(self):
    """Start the connection to the server."""

    self.reader, self.writer = await asyncio.open_connection(self.host, self.port)
    self.alive = True

stop() async

Stop the connection to the server.

Source code in src\dap\connection.py

async def stop(self):
    """Stop the connection to the server."""

    self.writer.close()
    await self.writer.wait_closed()
    self.alive = False

write(data) async

Write data to the server

Parameters:

Name	Type	Description	Default
data	bytes	The data to write to the server.	required

Source code in src\dap\connection.py

async def write(self, data: bytes):
    """Write data to the server

    Args:
        data (bytes): The data to write to the server.
    """

    self.writer.write(data)
    await self.writer.drain()

Connection

Connection to a debug adapter server.

This class is used to connect to a debug adapter server using threads. It provides methods to start, stop, read and write to the server.

Source code in src\dap\connection.py

class Connection:
    """Connection to a debug adapter server.

    This class is used to connect to a debug adapter server using threads.
    It provides methods to start, stop, read and write to the server."""

    def __init__(self, host="localhost", port=6789):
        self.alive = True
        self.host = host
        self.port = port

        self.out_queue = queue.Queue()

    def write(self, buf: bytes) -> None:
        """Write data to the server

        Args:
            buf (bytes): The data to write to the server.
        """

        self.sock.sendall(buf)

    def read(self) -> None:
        """Read data from the server

        Returns:
            bytes: The data read from the server.
        """

        buf = bytearray()
        while True:
            try:
                buf += self.out_queue.get(block=False)
            except queue.Empty:
                break

        if self.t_out.is_alive() and not buf:
            return None
        return bytes(buf)

    def start(self, *_) -> None:
        """Start the connection to the server."""

        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.sock.connect((self.host, self.port))

        self.t_out = Thread(target=self._process_output, daemon=True)
        self.t_out.start()

    def stop(self, *_) -> None:
        """Stop the connection to the server."""

        self.alive = False
        self.sock.close()

    def _process_output(self) -> None:
        while self.alive:
            try:
                data = self.sock.recv(1024)
            except ConnectionAbortedError:
                break
            except ConnectionResetError:
                break

            if not data:
                break

            self.out_queue.put(data)

read()

Read data from the server

Returns:

Name	Type	Description
bytes	None	The data read from the server.

Source code in src\dap\connection.py

def read(self) -> None:
    """Read data from the server

    Returns:
        bytes: The data read from the server.
    """

    buf = bytearray()
    while True:
        try:
            buf += self.out_queue.get(block=False)
        except queue.Empty:
            break

    if self.t_out.is_alive() and not buf:
        return None
    return bytes(buf)

start(*_)

Start the connection to the server.

Source code in src\dap\connection.py

def start(self, *_) -> None:
    """Start the connection to the server."""

    self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    self.sock.connect((self.host, self.port))

    self.t_out = Thread(target=self._process_output, daemon=True)
    self.t_out.start()

stop(*_)

Stop the connection to the server.

Source code in src\dap\connection.py

def stop(self, *_) -> None:
    """Stop the connection to the server."""

    self.alive = False
    self.sock.close()

write(buf)

Write data to the server

Parameters:

Name	Type	Description	Default
buf	bytes	The data to write to the server.	required

Source code in src\dap\connection.py

def write(self, buf: bytes) -> None:
    """Write data to the server

    Args:
        buf (bytes): The data to write to the server.
    """

    self.sock.sendall(buf)

Threaded Socket IO Client

Abstract Threaded server implementation of the DAP Client

It is meant to be used as a base class for creating a server. It handles the connection and the client. Following methods need to be implemented by the child class: - handle_message

Source code in src\dap\server.py

class ThreadedServer:
    """Abstract Threaded server implementation of the DAP Client

    It is meant to be used as a base class for creating a server. It handles the connection and the client.
    Following methods need to be implemented by the child class:
    - handle_message
    """

    def __init__(self, adapter_id: str, host="localhost", port=6789) -> None:
        """Initializes the server with the given adapter_id, host and port

        Args:
            adapter_id (str): The adapter id
            host (str, optional): The host to connect to. Defaults to "localhost".
            port (int, optional): The port to connect to. Defaults to 6789.
        """

        self.connection = Connection(host, port)
        self.connection.start()

        self.client = Client(adapter_id)
        self.running = False

    def start(self):
        """Starts the server"""

        self.running = True
        threading.Thread(target=self._run_loop, daemon=True).start()

    def stop(self):
        """Stops the server"""

        self.running = False
        self.client.terminate()
        self.connection.stop()

    def _run_loop(self):
        while self.running and self.connection.alive and self.run_single():
            ...

    def run_single(self):
        if s := self.client.send():
            self.connection.write(s)

        r = self.connection.read()
        if not r:
            return False

        for result in self.client.receive(r):
            self.handle_message(result)

        return True

    def handle_message(self, message):
        """Handles the message received from the client

        To be implemented by child classes
        Args:
            message (Any): The message to handle
        """

        print(type(message), flush=True)

__init__(adapter_id, host='localhost', port=6789)

Initializes the server with the given adapter_id, host and port

Parameters:

Name	Type	Description	Default
adapter_id	str	The adapter id	required
host	str	The host to connect to. Defaults to "localhost".	'localhost'
port	int	The port to connect to. Defaults to 6789.	6789

Source code in src\dap\server.py

def __init__(self, adapter_id: str, host="localhost", port=6789) -> None:
    """Initializes the server with the given adapter_id, host and port

    Args:
        adapter_id (str): The adapter id
        host (str, optional): The host to connect to. Defaults to "localhost".
        port (int, optional): The port to connect to. Defaults to 6789.
    """

    self.connection = Connection(host, port)
    self.connection.start()

    self.client = Client(adapter_id)
    self.running = False

handle_message(message)

Handles the message received from the client

To be implemented by child classes Args: message (Any): The message to handle

Source code in src\dap\server.py

def handle_message(self, message):
    """Handles the message received from the client

    To be implemented by child classes
    Args:
        message (Any): The message to handle
    """

    print(type(message), flush=True)

start()

Starts the server

Source code in src\dap\server.py

def start(self):
    """Starts the server"""

    self.running = True
    threading.Thread(target=self._run_loop, daemon=True).start()

stop()

Stops the server

Source code in src\dap\server.py

def stop(self):
    """Stops the server"""

    self.running = False
    self.client.terminate()
    self.connection.stop()

Asyncio Client

Abstract Asyncio-based DAP server.

Handles communication between the client and the adapter. Child classes should implement the following methods:

• handle_message: Handle a message from the client or adapter.

Example:

class MyServer(AsyncServer):
    def handle_message(self, message):
        print(message)

Source code in src\dap\asyncserver.py

class AsyncServer:
    """Abstract Asyncio-based DAP server.

    Handles communication between the client and the adapter. Child classes should
    implement the following methods:

    - handle_message: Handle a message from the client or adapter.

    Example:

    ```python
    class MyServer(AsyncServer):
        def handle_message(self, message):
            print(message)
    ```
    """

    def __init__(
        self, adapter_id: str, host: str = "localhost", port: int = 6789
    ) -> None:
        self.connection = AsyncConnection(host, port)
        self.client = Client(adapter_id)
        self.running = False
        self.loop: Optional[asyncio.AbstractEventLoop] = None

    async def start(self):
        """Start the server."""

        await self.connection.start()
        self.running = True
        self.loop = asyncio.get_running_loop()
        await self._run_loop()

    async def stop(self):
        """Stop the server."""

        self.running = False
        self.client.terminate()
        await self.connection.stop()

    async def _run_loop(self):
        while self.running and self.connection.alive:
            await self.run_single()
            await asyncio.sleep(0.1)

    async def run_single(self):
        s = self.client.send()
        if s:
            await self.connection.write(s)

        r = await self.connection.read()
        for event in self.client.receive(r):
            self.handle_message(event)

        return True

    def handle_message(self, message):
        """Handle a message from the client or adapter.

        To be implemented by subclasses.
        """

        print(type(message), flush=True)

handle_message(message)

Handle a message from the client or adapter.

To be implemented by subclasses.

Source code in src\dap\asyncserver.py

def handle_message(self, message):
    """Handle a message from the client or adapter.

    To be implemented by subclasses.
    """

    print(type(message), flush=True)

start() async

Start the server.

Source code in src\dap\asyncserver.py

async def start(self):
    """Start the server."""

    await self.connection.start()
    self.running = True
    self.loop = asyncio.get_running_loop()
    await self._run_loop()

stop() async

Stop the server.

Source code in src\dap\asyncserver.py

async def stop(self):
    """Stop the server."""

    self.running = False
    self.client.terminate()
    await self.connection.stop()