"""MCP Debug Server - Main server implementation."""
import asyncio
import sys
import os
from typing import Any, Optional
from mcp.server import Server
from mcp.types import Tool, TextContent
import mcp.server.stdio
# Handle both package import and direct script execution
if __name__ == "__main__" or not __package__:
# Add parent directory to path for direct execution
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
from mcp_debugserver.debugger_manager import DebuggerManager
from mcp_debugserver.windbg_commands import WinDbgCommands, WinDbgParser
else:
from .debugger_manager import DebuggerManager
from .windbg_commands import WinDbgCommands, WinDbgParser
class DebugServer:
"""MCP Server for Windows debugging with WinDbg/CDB."""
def __init__(self):
self.server = Server("mcp-debugserver")
# Auto-detect debugger paths
cdb_path, dbgsrv_path = self._find_debugger_paths()
self.manager = DebuggerManager(dbgsrv_path=dbgsrv_path, cdb_path=cdb_path)
self.commands = WinDbgCommands()
self.parser = WinDbgParser()
self._setup_handlers()
def _find_debugger_paths(self):
"""
Find cdb.exe and dbgsrv.exe in common locations.
Checks the 'arch' environment variable to determine architecture preference:
- arch=x86: Prefer x86 debuggers
- arch=x64 or not set: Prefer x64 debuggers (default)
Returns:
Tuple of (cdb_path, dbgsrv_path)
"""
# Check for architecture preference from environment variable
arch_preference = os.environ.get('arch', 'x64').lower()
# Initialize paths
cdb_path = None
dbgsrv_path = None
# Check common Windows SDK locations based on architecture preference
if arch_preference == 'x86':
common_paths = [
r"C:\Program Files (x86)\Windows Kits\10\Debuggers\x86",
r"C:\Program Files\Windows Kits\10\Debuggers\x86",
r"C:\Program Files (x86)\Windows Kits\10\Debuggers\x64",
r"C:\Program Files\Windows Kits\10\Debuggers\x64",
]
else: # Default to x64
common_paths = [
r"C:\Program Files (x86)\Windows Kits\10\Debuggers\x64",
r"C:\Program Files\Windows Kits\10\Debuggers\x64",
r"C:\Program Files (x86)\Windows Kits\10\Debuggers\x86",
r"C:\Program Files\Windows Kits\10\Debuggers\x86",
]
for base_path in common_paths:
if not cdb_path:
test_cdb = os.path.join(base_path, "cdb.exe")
if os.path.exists(test_cdb):
cdb_path = test_cdb
if not dbgsrv_path:
test_dbgsrv = os.path.join(base_path, "dbgsrv.exe")
if os.path.exists(test_dbgsrv):
dbgsrv_path = test_dbgsrv
if cdb_path and dbgsrv_path:
break
# Use defaults if still not found (will fail later with helpful error)
if not cdb_path:
cdb_path = r"C:\Program Files (x86)\Windows Kits\10\Debuggers\x64\cdb.exe"
if not dbgsrv_path:
dbgsrv_path = r"C:\Program Files (x86)\Windows Kits\10\Debuggers\x64\dbgsrv.exe"
return cdb_path, dbgsrv_path
def _setup_handlers(self):
"""Setup MCP protocol handlers."""
@self.server.list_tools()
async def list_tools() -> list[Tool]:
"""List all available debugging tools."""
return [
# Server lifecycle management
Tool(
name="start_debug_server",
description="Start CDB in server mode on specified TCP port, optionally attached to a process",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port for CDB server (default: 5005)",
"default": 5005
},
"pid": {
"type": "integer",
"description": "Process ID to attach to immediately (optional)"
},
"process_name": {
"type": "string",
"description": "Process name to attach to immediately (optional, e.g., 'notepad.exe')"
}
}
}
),
Tool(
name="stop_debug_server",
description="Stop debug server running on specified port",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug server to stop"
}
},
"required": ["port"]
}
),
Tool(
name="connect_debugger",
description="Connect CDB client to running CDB server session",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of CDB server"
},
"server_address": {
"type": "string",
"description": "Server address (default: localhost)",
"default": "localhost"
}
},
"required": ["port"]
}
),
Tool(
name="disconnect_debugger",
description="Disconnect debugger client and close session",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
}
},
"required": ["port"]
}
),
# Process attachment
Tool(
name="attach_process",
description="Attach debugger to a process by PID or name (optional if already attached during server start)",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
},
"pid": {
"type": "integer",
"description": "Process ID to attach to"
},
"process_name": {
"type": "string",
"description": "Process name to attach to (e.g., 'notepad.exe')"
}
},
"required": ["port"]
}
),
Tool(
name="detach_process",
description="Detach debugger from current process",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
}
},
"required": ["port"]
}
),
# Execution control
Tool(
name="break_execution",
description="Break into debugger (pause process execution)",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
}
},
"required": ["port"]
}
),
Tool(
name="resume_execution",
description="Resume process execution (go/continue)",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
}
},
"required": ["port"]
}
),
Tool(
name="step_into",
description="Step into next instruction (trace)",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
}
},
"required": ["port"]
}
),
Tool(
name="step_over",
description="Step over next instruction",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
}
},
"required": ["port"]
}
),
Tool(
name="step_out",
description="Step out of current function",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
}
},
"required": ["port"]
}
),
# Breakpoints
Tool(
name="set_breakpoint",
description="Set breakpoint at address or symbol",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
},
"address_or_symbol": {
"type": "string",
"description": "Memory address (hex) or symbol name (e.g., 'kernel32!CreateFileW', '0x7fff1234')"
},
"condition": {
"type": "string",
"description": "Optional condition for conditional breakpoint"
},
"command": {
"type": "string",
"description": "Optional command to execute when breakpoint hits"
}
},
"required": ["port", "address_or_symbol"]
}
),
Tool(
name="remove_breakpoint",
description="Remove breakpoint by ID",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
},
"breakpoint_id": {
"type": "integer",
"description": "Breakpoint ID number"
}
},
"required": ["port", "breakpoint_id"]
}
),
Tool(
name="list_breakpoints",
description="List all active breakpoints",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
}
},
"required": ["port"]
}
),
Tool(
name="enable_breakpoint",
description="Enable a disabled breakpoint",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
},
"breakpoint_id": {
"type": "integer",
"description": "Breakpoint ID to enable"
}
},
"required": ["port", "breakpoint_id"]
}
),
Tool(
name="disable_breakpoint",
description="Disable a breakpoint without removing it",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
},
"breakpoint_id": {
"type": "integer",
"description": "Breakpoint ID to disable"
}
},
"required": ["port", "breakpoint_id"]
}
),
# Memory operations
Tool(
name="read_memory",
description="Read memory at specified address",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
},
"address": {
"type": "string",
"description": "Memory address in hex (e.g., '0x7fff1234')"
},
"length": {
"type": "integer",
"description": "Number of units to read",
"default": 16
},
"format": {
"type": "string",
"description": "Display format: b(bytes), w(words), d(dwords), q(qwords), a(ASCII), u(Unicode)",
"enum": ["b", "w", "d", "q", "a", "u", "f", "D"],
"default": "b"
}
},
"required": ["port", "address"]
}
),
Tool(
name="write_memory",
description="Write data to memory at specified address",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
},
"address": {
"type": "string",
"description": "Memory address in hex"
},
"values": {
"type": "array",
"items": {"type": "integer"},
"description": "Array of integer values to write"
},
"format": {
"type": "string",
"description": "Data format: b(byte), w(word), d(dword), q(qword)",
"enum": ["b", "w", "d", "q"],
"default": "b"
}
},
"required": ["port", "address", "values"]
}
),
Tool(
name="search_memory",
description="Search memory for a pattern",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
},
"start_address": {
"type": "string",
"description": "Start address for search"
},
"end_address": {
"type": "string",
"description": "End address or length (e.g., 'L1000' for 0x1000 bytes)"
},
"pattern": {
"type": "string",
"description": "Pattern to search (hex bytes, ASCII, or Unicode)"
},
"data_type": {
"type": "string",
"description": "Data type: b(byte), w(word), d(dword), q(qword), a(ASCII), u(Unicode)",
"default": "b"
}
},
"required": ["port", "start_address", "end_address", "pattern"]
}
),
# Registers
Tool(
name="read_registers",
description="Read CPU register values",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
},
"register_name": {
"type": "string",
"description": "Specific register name (optional, reads all if omitted)"
}
},
"required": ["port"]
}
),
Tool(
name="write_register",
description="Write value to a CPU register",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
},
"register_name": {
"type": "string",
"description": "Register name (e.g., 'rax', 'eip', 'rbp')"
},
"value": {
"type": "string",
"description": "Value to write (hex string)"
}
},
"required": ["port", "register_name", "value"]
}
),
# Call stack and threads
Tool(
name="get_callstack",
description="Get current call stack (stack trace)",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
},
"frames": {
"type": "integer",
"description": "Number of frames to display (optional)"
},
"verbose": {
"type": "boolean",
"description": "Include detailed information",
"default": False
}
},
"required": ["port"]
}
),
Tool(
name="list_threads",
description="List all threads in target process",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
}
},
"required": ["port"]
}
),
Tool(
name="switch_thread",
description="Switch debugger context to specific thread",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
},
"thread_id": {
"type": "integer",
"description": "Thread ID to switch to"
}
},
"required": ["port", "thread_id"]
}
),
# Modules and symbols
Tool(
name="list_modules",
description="List all loaded modules/DLLs",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
},
"verbose": {
"type": "boolean",
"description": "Include detailed module information",
"default": False
}
},
"required": ["port"]
}
),
Tool(
name="module_info",
description="Get detailed information about a specific module",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
},
"module_name": {
"type": "string",
"description": "Module name (e.g., 'kernel32', 'ntdll')"
}
},
"required": ["port", "module_name"]
}
),
Tool(
name="evaluate_expression",
description="Evaluate expression or resolve symbol address",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
},
"expression": {
"type": "string",
"description": "Expression to evaluate (address, symbol, calculation)"
}
},
"required": ["port", "expression"]
}
),
Tool(
name="display_symbol",
description="Display symbol information (resolve wildcards)",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
},
"symbol_pattern": {
"type": "string",
"description": "Symbol pattern (supports wildcards, e.g., 'kernel32!Create*')"
}
},
"required": ["port", "symbol_pattern"]
}
),
Tool(
name="reload_symbols",
description="Reload symbols for all modules",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
}
},
"required": ["port"]
}
),
Tool(
name="set_symbol_path",
description="Set symbol search path",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
},
"path": {
"type": "string",
"description": "Symbol path (can include srv* for symbol server)"
},
"append": {
"type": "boolean",
"description": "Append to existing path instead of replacing",
"default": False
}
},
"required": ["port", "path"]
}
),
# Disassembly
Tool(
name="disassemble",
description="Disassemble code at address",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
},
"address": {
"type": "string",
"description": "Start address for disassembly"
},
"length": {
"type": "integer",
"description": "Number of instructions to disassemble"
}
},
"required": ["port", "address"]
}
),
# Process and exception info
Tool(
name="get_process_info",
description="Get current process information",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
}
},
"required": ["port"]
}
),
Tool(
name="get_exception_info",
description="Get last exception/event information",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
}
},
"required": ["port"]
}
),
# Raw command execution
Tool(
name="execute_command",
description="Execute any raw WinDbg command directly",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
},
"command": {
"type": "string",
"description": "WinDbg command to execute"
}
},
"required": ["port", "command"]
}
),
# Exception monitoring
Tool(
name="enable_exception_monitoring",
description="Enable background exception monitoring with auto-capture",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
},
"auto_freeze_on_exception": {
"type": "boolean",
"description": "Freeze all threads when exception occurs (default: true)",
"default": True
},
"capture_full_state": {
"type": "boolean",
"description": "Capture registers, callstack, etc. (default: true)",
"default": True
},
"generate_minidump": {
"type": "boolean",
"description": "Generate minidump file on exception (default: true)",
"default": True
},
"minidump_path": {
"type": "string",
"description": "Path for minidump files (default: C:\\dumps)"
}
},
"required": ["port"]
}
),
Tool(
name="get_last_exception_captured",
description="Get the last captured exception data (even after thread termination)",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
}
},
"required": ["port"]
}
),
Tool(
name="create_minidump",
description="Generate minidump on demand",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
},
"output_path": {
"type": "string",
"description": "Path for the dump file"
},
"dump_type": {
"type": "string",
"description": "Dump type: 'mini', 'heap', or 'full' (default)",
"default": "full"
}
},
"required": ["port", "output_path"]
}
),
Tool(
name="set_exception_break",
description="Set exception breakpoint with auto-capture",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
},
"exception_code": {
"type": "string",
"description": "Exception code ('av', 'c0000005', '*' for all)"
},
"break_type": {
"type": "string",
"description": "Break on 'first' or 'second' chance (default: first)",
"default": "first"
},
"auto_capture_state": {
"type": "boolean",
"description": "Auto-capture state on break (default: true)",
"default": True
}
},
"required": ["port", "exception_code"]
}
),
Tool(
name="resume_with_monitoring",
description="Resume execution with exception monitoring",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
},
"timeout": {
"type": "integer",
"description": "Seconds to wait for exception (None = no timeout)"
},
"stop_on_exception": {
"type": "boolean",
"description": "Auto-stop when exception occurs (default: true)",
"default": True
}
},
"required": ["port"]
}
),
# Hardware breakpoints
Tool(
name="set_hardware_breakpoint",
description="Set hardware breakpoint using CPU debug registers (fast, non-intrusive, works with fast threads)",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
},
"address": {
"type": "string",
"description": "Memory address in hex (e.g., '0x140001234')"
},
"hw_type": {
"type": "string",
"description": "Breakpoint type: 'execute', 'write', 'readwrite', or 'io'",
"default": "execute"
},
"size": {
"type": "integer",
"description": "Size of watched region in bytes (1, 2, 4, or 8). Only applies to data breakpoints.",
"default": 1
},
"condition": {
"type": "string",
"description": "Optional WinDbg condition expression (e.g., 'rcx > 0x1000')"
},
"thread_id": {
"type": "integer",
"description": "Apply only to specific thread (None = all threads)"
}
},
"required": ["port", "address"]
}
),
Tool(
name="remove_hardware_breakpoint",
description="Remove a hardware breakpoint",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
},
"breakpoint_id": {
"type": "integer",
"description": "Hardware breakpoint ID to remove"
}
},
"required": ["port", "breakpoint_id"]
}
),
Tool(
name="list_hardware_breakpoints",
description="List all active hardware breakpoints with hit counts and status",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
}
},
"required": ["port"]
}
),
Tool(
name="toggle_hardware_breakpoint",
description="Enable or disable a hardware breakpoint without removing it",
inputSchema={
"type": "object",
"properties": {
"port": {
"type": "integer",
"description": "TCP port of debug session"
},
"breakpoint_id": {
"type": "integer",
"description": "Hardware breakpoint ID"
},
"enabled": {
"type": "boolean",
"description": "True to enable, False to disable"
}
},
"required": ["port", "breakpoint_id", "enabled"]
}
),
]
@self.server.call_tool()
async def call_tool(name: str, arguments: Any) -> list[TextContent]:
"""Handle tool execution."""
try:
result = await self._handle_tool(name, arguments)
return [TextContent(type="text", text=str(result))]
except Exception as e:
return [TextContent(type="text", text=f"Error: {str(e)}")]
async def _handle_tool(self, name: str, args: dict) -> dict:
"""Route tool calls to appropriate handlers."""
# Server lifecycle
if name == "start_debug_server":
return await self.manager.start_debug_server(
args.get("port", 5005),
args.get("pid"),
args.get("process_name")
)
elif name == "stop_debug_server":
return await self.manager.stop_debug_server(args["port"])
elif name == "connect_debugger":
return await self.manager.connect_to_server(
args["port"],
args.get("server_address", "localhost")
)
elif name == "disconnect_debugger":
return await self.manager.stop_debug_server(args["port"])
# Process attachment
elif name == "attach_process":
return await self.manager.attach_to_process(
args["port"],
args.get("pid"),
args.get("process_name")
)
elif name == "detach_process":
return await self.manager.detach_from_process(args["port"])
# Execution control
elif name == "break_execution":
cmd = self.commands.break_execution()
return await self.manager.execute_command(args["port"], cmd)
elif name == "resume_execution":
cmd = self.commands.resume_execution()
return await self.manager.execute_command(args["port"], cmd)
elif name == "step_into":
cmd = self.commands.step_into()
return await self.manager.execute_command(args["port"], cmd)
elif name == "step_over":
cmd = self.commands.step_over()
return await self.manager.execute_command(args["port"], cmd)
elif name == "step_out":
cmd = self.commands.step_out()
return await self.manager.execute_command(args["port"], cmd)
# Breakpoints
elif name == "set_breakpoint":
if args.get("command"):
cmd = self.commands.set_breakpoint_with_command(
args["address_or_symbol"],
args["command"]
)
else:
cmd = self.commands.set_breakpoint(
args["address_or_symbol"],
args.get("condition")
)
return await self.manager.execute_command(args["port"], cmd)
elif name == "remove_breakpoint":
cmd = self.commands.remove_breakpoint(args["breakpoint_id"])
return await self.manager.execute_command(args["port"], cmd)
elif name == "list_breakpoints":
cmd = self.commands.list_breakpoints()
result = await self.manager.execute_command(args["port"], cmd)
if result.get("status") == "success":
result["breakpoints"] = self.parser.parse_breakpoints(result["output"])
return result
elif name == "enable_breakpoint":
cmd = self.commands.enable_breakpoint(args["breakpoint_id"])
return await self.manager.execute_command(args["port"], cmd)
elif name == "disable_breakpoint":
cmd = self.commands.disable_breakpoint(args["breakpoint_id"])
return await self.manager.execute_command(args["port"], cmd)
# Memory operations
elif name == "read_memory":
cmd = self.commands.read_memory(
args["address"],
args.get("length", 16),
args.get("format", "b")
)
return await self.manager.execute_command(args["port"], cmd)
elif name == "write_memory":
cmd = self.commands.write_memory(
args["address"],
args["values"],
args.get("format", "b")
)
return await self.manager.execute_command(args["port"], cmd)
elif name == "search_memory":
cmd = self.commands.search_memory(
args["start_address"],
args["end_address"],
args["pattern"],
args.get("data_type", "b")
)
return await self.manager.execute_command(args["port"], cmd)
# Registers
elif name == "read_registers":
if args.get("register_name"):
cmd = self.commands.read_register(args["register_name"])
else:
cmd = self.commands.read_registers()
result = await self.manager.execute_command(args["port"], cmd)
if result.get("status") == "success":
result["registers"] = self.parser.parse_registers(result["output"])
return result
elif name == "write_register":
cmd = self.commands.write_register(args["register_name"], args["value"])
return await self.manager.execute_command(args["port"], cmd)
# Call stack and threads
elif name == "get_callstack":
if args.get("verbose"):
cmd = self.commands.get_callstack_verbose()
else:
cmd = self.commands.get_callstack(args.get("frames"))
result = await self.manager.execute_command(args["port"], cmd)
if result.get("status") == "success":
result["frames"] = self.parser.parse_callstack(result["output"])
return result
elif name == "list_threads":
cmd = self.commands.list_threads()
result = await self.manager.execute_command(args["port"], cmd)
if result.get("status") == "success":
result["threads"] = self.parser.parse_threads(result["output"])
return result
elif name == "switch_thread":
cmd = self.commands.switch_thread(args["thread_id"])
return await self.manager.execute_command(args["port"], cmd)
# Modules and symbols
elif name == "list_modules":
if args.get("verbose"):
cmd = self.commands.list_modules_verbose()
else:
cmd = self.commands.list_modules()
result = await self.manager.execute_command(args["port"], cmd)
if result.get("status") == "success":
result["modules"] = self.parser.parse_modules(result["output"])
return result
elif name == "module_info":
cmd = self.commands.module_info(args["module_name"])
return await self.manager.execute_command(args["port"], cmd)
elif name == "evaluate_expression":
cmd = self.commands.evaluate_expression(args["expression"])
result = await self.manager.execute_command(args["port"], cmd)
if result.get("status") == "success":
result["value"] = self.parser.extract_value(result["output"])
return result
elif name == "display_symbol":
cmd = self.commands.display_symbol(args["symbol_pattern"])
return await self.manager.execute_command(args["port"], cmd)
elif name == "reload_symbols":
cmd = self.commands.reload_symbols()
return await self.manager.execute_command(args["port"], cmd)
elif name == "set_symbol_path":
if args.get("append"):
cmd = self.commands.add_symbol_path(args["path"])
else:
cmd = self.commands.set_symbol_path(args["path"])
return await self.manager.execute_command(args["port"], cmd)
# Disassembly
elif name == "disassemble":
cmd = self.commands.disassemble(
args["address"],
args.get("length")
)
return await self.manager.execute_command(args["port"], cmd)
# Process and exception info
elif name == "get_process_info":
cmd = self.commands.get_process_info()
return await self.manager.execute_command(args["port"], cmd)
elif name == "get_exception_info":
cmd = self.commands.get_exception_info()
return await self.manager.execute_command(args["port"], cmd)
# Raw command execution
elif name == "execute_command":
return await self.manager.execute_command(args["port"], args["command"])
# Exception monitoring
elif name == "enable_exception_monitoring":
return await self.manager.enable_exception_monitoring(
args["port"],
args.get("auto_freeze_on_exception", True),
args.get("capture_full_state", True),
args.get("generate_minidump", True),
args.get("minidump_path")
)
elif name == "get_last_exception_captured":
return await self.manager.get_last_exception_captured(args["port"])
elif name == "create_minidump":
return await self.manager.create_minidump(
args["port"],
args["output_path"],
args.get("dump_type", "full")
)
elif name == "set_exception_break":
return await self.manager.set_exception_break(
args["port"],
args["exception_code"],
args.get("break_type", "first"),
args.get("auto_capture_state", True)
)
elif name == "resume_with_monitoring":
return await self.manager.resume_with_monitoring(
args["port"],
args.get("timeout"),
args.get("stop_on_exception", True)
)
# Hardware breakpoints
elif name == "set_hardware_breakpoint":
return await self.manager.set_hardware_breakpoint(
args["port"],
args["address"],
args.get("hw_type", "execute"),
args.get("size", 1),
args.get("condition"),
args.get("thread_id")
)
elif name == "remove_hardware_breakpoint":
return await self.manager.remove_hardware_breakpoint(
args["port"],
args["breakpoint_id"]
)
elif name == "list_hardware_breakpoints":
return await self.manager.list_hardware_breakpoints(args["port"])
elif name == "toggle_hardware_breakpoint":
return await self.manager.toggle_hardware_breakpoint(
args["port"],
args["breakpoint_id"],
args["enabled"]
)
else:
return {"status": "error", "message": f"Unknown tool: {name}"}
async def run(self):
"""Run the MCP server."""
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
await self.server.run(
read_stream,
write_stream,
self.server.create_initialization_options()
)
async def main():
"""Main entry point."""
server = DebugServer()
try:
await server.run()
finally:
await server.manager.cleanup()
def run():
"""Synchronous entry point for scripts."""
asyncio.run(main())
if __name__ == "__main__":
run()
