run_drgn_command
Execute drgn Python commands to analyze Linux crash dumps, enabling interactive debugging of system failures through kernel variable inspection and thread analysis.
Instructions
Runs drgn Python code (e.g., prog.crashed_thread(), prog['variable']).
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| command | Yes | ||
| session_id | No | ||
| truncate | No |
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)}"
- 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
- 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)