import os
import sys
import ast
import json
import argparse
import http.client
from urllib.parse import urlparse
from mcp.server.fastmcp import FastMCP
# 设置日志级别以兼容 Cline
mcp = FastMCP("windbg-gui-mcp", log_level="ERROR")
jsonrpc_request_id = 1
windbg_host = "127.0.0.1"
windbg_port = 13338
def make_jsonrpc_request(method: str, *params):
"""向 WinDbg 插件发送 JSON-RPC 请求"""
global jsonrpc_request_id, windbg_host, windbg_port
conn = http.client.HTTPConnection(windbg_host, windbg_port)
request = {
"jsonrpc": "2.0",
"method": method,
"params": list(params),
"id": jsonrpc_request_id,
}
jsonrpc_request_id += 1
try:
conn.request("POST", "/windbg", json.dumps(request), {
"Content-Type": "application/json"
})
response = conn.getresponse()
data = json.loads(response.read().decode())
if "error" in data:
error = data["error"]
code = error["code"]
message = error["message"]
pretty = f"JSON-RPC error {code}: {message}"
if "data" in error:
pretty += "\n" + error["data"]
raise Exception(pretty)
result = data["result"]
# LLM 对空响应处理不好，返回成功消息
if result is None:
result = "success"
return result
except Exception:
raise
finally:
conn.close()
@mcp.tool()
def check_connection() -> str:
"""检查 WinDbg 插件是否正在运行"""
try:
info = make_jsonrpc_request("get_debugger_info")
arch = info.get("target_arch", "unknown")
is_kernel = info.get("is_kernel_mode", False)
mode = "kernel" if is_kernel else "user"
return f"Successfully connected to WinDbg ({mode} mode, {arch} architecture)"
except Exception as e:
return f"Failed to connect to WinDbg! Make sure you:\n1. Loaded the PyKD extension in WinDbg\n2. Ran: .load pykd\n3. Executed the windbg_plugin.py script\n\nError: {str(e)}"
@mcp.tool()
def get_debugger_info() -> dict:
"""获取当前调试器会话的信息"""
return make_jsonrpc_request("get_debugger_info")
@mcp.tool()
def execute_command(command: str) -> str:
"""执行 WinDbg 命令并返回输出"""
return make_jsonrpc_request("execute_command", command)
@mcp.tool()
def get_stack_trace() -> list:
"""获取当前调用堆栈"""
return make_jsonrpc_request("get_stack_trace")
@mcp.tool()
def get_registers() -> dict:
"""获取所有 CPU 寄存器"""
return make_jsonrpc_request("get_registers")
@mcp.tool()
def get_modules() -> list:
"""获取已加载模块的信息"""
return make_jsonrpc_request("get_modules")
@mcp.tool()
def read_memory(address: str, size: int) -> str:
"""读取指定地址的内存
Args:
address: 十六进制格式的内存地址 (例如 '0x1000')
size: 要读取的字节数
"""
return make_jsonrpc_request("read_memory", address, size)
@mcp.tool()
def analyze_exception() -> str:
"""对当前异常运行 !analyze -v"""
return make_jsonrpc_request("analyze_exception")
@mcp.tool()
def get_threads() -> list:
"""获取所有线程的信息"""
return make_jsonrpc_request("get_threads")
def get_python_executable():
"""获取 Python 可执行文件的路径"""
venv = os.environ.get("VIRTUAL_ENV")
if venv:
if sys.platform == "win32":
python = os.path.join(venv, "Scripts", "python.exe")
else:
python = os.path.join(venv, "bin", "python3")
if os.path.exists(python):
return python
for path in sys.path:
if sys.platform == "win32":
path = path.replace("/", "\\")
split = path.split(os.sep)
if split[-1].endswith(".zip"):
path = os.path.dirname(path)
if sys.platform == "win32":
python_executable = os.path.join(path, "python.exe")
else:
python_executable = os.path.join(path, "..", "bin", "python3")
python_executable = os.path.abspath(python_executable)
if os.path.exists(python_executable):
return python_executable
return sys.executable
def print_mcp_config():
"""打印 MCP 客户端配置"""
mcp_config = {
"command": get_python_executable(),
"args": [
"-m",
"windbg_gui_mcp",
],
"timeout": 1800,
"disabled": False,
}
print(json.dumps({
"mcpServers": {
mcp.name: mcp_config
}
}, indent=2)
)
def main():
global windbg_host, windbg_port
parser = argparse.ArgumentParser(description="WinDbg GUI MCP Server")
parser.add_argument("--transport", type=str, default="stdio", help="MCP transport protocol to use (stdio or http://127.0.0.1:8744)")
parser.add_argument("--windbg-rpc", type=str, default=f"http://{windbg_host}:{windbg_port}", help=f"WinDbg RPC server to use (default: http://{windbg_host}:{windbg_port})")
parser.add_argument("--config", action="store_true", help="Generate MCP config JSON")
args = parser.parse_args()
if args.config:
print_mcp_config()
return
# Parse WinDbg RPC server argument
windbg_rpc = urlparse(args.windbg_rpc)
if windbg_rpc.hostname is None or windbg_rpc.port is None:
raise Exception(f"Invalid WinDbg RPC server: {args.windbg_rpc}")
windbg_host = windbg_rpc.hostname
windbg_port = windbg_rpc.port
try:
if args.transport == "stdio":
mcp.run(transport="stdio")
else:
url = urlparse(args.transport)
if url.hostname is None or url.port is None:
raise Exception(f"Invalid transport URL: {args.transport}")
mcp.settings.host = url.hostname
mcp.settings.port = url.port
print(f"MCP Server available at http://{mcp.settings.host}:{mcp.settings.port}/sse")
mcp.settings.log_level = "INFO"
mcp.run(transport="sse")
except KeyboardInterrupt:
pass
if __name__ == "__main__":
main()
