Crash MCP Server

Instructions

Implementation Reference

• src/crash_mcp/server.py:114-133 (handler)

  Primary handler function for the 'run_drgn_command' MCP tool. Retrieves the session by ID (or uses the last one), checks if active, and delegates execution to UnifiedSession.execute_command with a 'drgn:' prefix.

  @mcp.tool()
  def run_drgn_command(command: str, session_id: Optional[str] = None, truncate: bool = True) -> str:
      """Runs drgn Python code (e.g., prog.crashed_thread(), prog['variable'])."""
      target_id = session_id or last_session_id
      
      if not target_id:
          return "Error: No session specified and no active default session."
      
      if target_id not in sessions:
          return f"Error: Session ID {target_id} not found."
      
      session = sessions[target_id]
      if not session.is_active():
          del sessions[target_id]
          return "Error: Session is no longer active."
          
      try:
          return session.execute_command(f"drgn:{command}", truncate=truncate)
      except Exception as e:
          return f"Error executing command: {str(e)}"

• src/crash_mcp/unified_session.py:98-102 (helper)

  Helper method in UnifiedSession that routes drgn commands to the DrgnSession instance.

  def _exec_drgn(self, cmd: str, timeout: int, truncate: bool) -> str:
      if not self.drgn_session or not self.drgn_session.is_active():
          return "Error: Drgn engine is not active."
      return self.drgn_session.execute_command(cmd, timeout, truncate)

• src/crash_mcp/drgn_session.py:83-134 (helper)

  Core execution logic in DrgnSession: spawns and interacts with the 'drgn' process using pexpect, sends the Python command, captures and processes output, applies truncation if enabled.

  def execute_command(self, command: str, timeout: int = 60, truncate: bool = True) -> str:
      """
      Executes a python command in the drgn session and returns the output.
      """
      if not self._process or not self._process.isalive():
          raise RuntimeError("Drgn session is not active")
  
      logger.debug(f"Executing drgn command: {command}")
      
      if command.strip() in ['quit()', 'exit()']:
          self.close()
          return "Session closed"
  
      try:
          # Send the command
          self._process.sendline(command)
          
          # Wait for prompt
          self._process.expect(self.PROMPT, timeout=timeout)
          
          # content before the prompt is the command output (plus the command echo)
          raw_output = self._process.before
          
          # Clean up the output
          # 1. Remove the command echo (first line usually)
          lines = raw_output.splitlines()
          if lines and command.strip() in lines[0]:
              lines = lines[1:]
          
          output = "\n".join(lines).strip()
          
          if not truncate:
              return output
          
          # Smart Truncation
          MAX_LEN = 16384 
          
          if len(output) > MAX_LEN:
              removed_chars = len(output) - MAX_LEN
              logger.warning(f"Output truncated. Original length: {len(output)}. Removed {removed_chars} chars.")
              
              HEAD_LEN = 4096
              TAIL_LEN = MAX_LEN - HEAD_LEN
              
              output = (
                  output[:HEAD_LEN] + 
                  f"\n\n... [Output truncated by Drgn MCP ({removed_chars} characters skipped)] ...\n\n" + 
                  output[-TAIL_LEN:]
              )
          
          return output

• src/crash_mcp/unified_session.py:57-97 (helper)

  UnifiedSession.execute_command method handles command routing based on 'drgn:' prefix (used by the tool handler), delegating to DrgnSession for drgn commands.

  def execute_command(self, command: str, timeout: int = 60, truncate: bool = True) -> str:
      """
      Routes the command to the appropriate engine.
      """
      cmd = command.strip()
      
      # Explicit routing prefix
      if cmd.startswith("drgn:"):
          return self._exec_drgn(cmd[5:].strip(), timeout, truncate)
      elif cmd.startswith("crash:"):
          return self._exec_crash(cmd[6:].strip(), timeout, truncate)
  
          
      # Heuristic Routing
      # Drgn is Python-based. Look for python syntax or known objects.
      is_drgn = False
      
      # Indicators of python/drgn code
      drgn_indicators = ['prog', 'libkdumpfile', 'find_task', '(', ')', '=', '.', '[', ']', '"', "'"]
      # Indicators of crash commands (simple words, known cmds)
      crash_cmds = ['sys', 'bt', 'ps', 'log', 'mount', 'net', 'dev', 'files', 'help', 'set', 'extend']
      
      first_word = cmd.split()[0] if cmd else ""
      
      if first_word in crash_cmds:
           is_drgn = False
      elif any(x in cmd for x in ['=', '(', '.', '[']):
          # Assignment, function call, attributes -> likely Python
          is_drgn = True
      elif first_word in ['prog', 'task', 'thread']:
          is_drgn = True
      else:
          # Default fallback to crash if ambiguous? Or try both?
          # Defaulting to crash is safer for system admins used to crash.
          is_drgn = False
          
      if is_drgn:
          return self._exec_drgn(cmd, timeout, truncate)
      else:
          return self._exec_crash(cmd, timeout, truncate)