Method send_request

ms_agent/tools/code_server/lsp_code_server.py:55–99

Send a JSON-RPC request to the LSP server

(self, method: str, params: dict = None)

                self.initialized = False

    async def send_request(self, method: str, params: dict = None) -> dict:
        """Send a JSON-RPC request to the LSP server"""
        if not self.process or not self.stdin or not self.stdout:
            raise RuntimeError('LSP server not started')

        self.message_id += 1
        request_id = self.message_id
        request = {
            'jsonrpc': '2.0',
            'id': request_id,
            'method': method,
            'params': params or {}
        }

        content = json.dumps(request)
        message = f'Content-Length: {len(content)}\r\n\r\n{content}'

        try:
            self.stdin.write(message.encode('utf-8'))
            await self.stdin.drain()

            max_retries = 20
            for _ in range(max_retries):
                msg = await self._read_message()

                # Check if it's the response we're waiting for
                if 'id' in msg and msg['id'] == request_id:
                    return msg

                # It's a notification (no id) or response for different request
                # Log and continue reading
                if 'method' in msg:
                    logger.debug(
                        f"Received notification during request: {msg.get('method')}"
                    )
                    continue

            logger.warning(
                f'No response received for request {request_id} after {max_retries} attempts'
            )
            return {'error': 'No response received'}

        except Exception as e:
            logger.error(f'Error sending LSP request: {e}')
            return {'error': str(e)}

    async def send_notification(self, method: str, params: dict = None):
        """Send a JSON-RPC notification to the LSP server"""