1578 lines
64 KiB
Python
1578 lines
64 KiB
Python
import os
|
||
import sys
|
||
import subprocess
|
||
import asyncio
|
||
from pathlib import Path
|
||
from fastapi import FastAPI, Request
|
||
from fastapi.responses import JSONResponse
|
||
from pydantic import BaseModel
|
||
from typing import Optional, Dict
|
||
import json
|
||
import platform
|
||
import re
|
||
from datetime import datetime
|
||
import tempfile
|
||
import uuid
|
||
import shlex
|
||
import threading
|
||
import time
|
||
import paramiko
|
||
import base64
|
||
import secrets
|
||
|
||
# ========== 禁用 __pycache__ ==========
|
||
os.environ['PYTHONDONTWRITEBYTECODE'] = '1'
|
||
sys.dont_write_bytecode = True
|
||
|
||
# ========== 命令行确认/输入支持 ==========
|
||
def is_windows():
|
||
return platform.system() == "Windows"
|
||
|
||
MB_YESNO = 0x04
|
||
MB_YESNOCANCEL = 0x03
|
||
MB_OKCANCEL = 0x01
|
||
MB_OK = 0x00
|
||
MB_ICONQUESTION = 0x20
|
||
MB_ICONWARNING = 0x30
|
||
MB_ICONINFORMATION = 0x40
|
||
MB_ICONERROR = 0x10
|
||
|
||
IDYES = 6
|
||
IDNO = 7
|
||
IDCANCEL = 2
|
||
IDOK = 1
|
||
|
||
def show_confirm_dialog(title: str, message: str, buttons: int = MB_YESNO, icon: int = MB_ICONQUESTION) -> int:
|
||
"""命令行确认（纯 input，不使用 GUI 弹窗）"""
|
||
print(f"\n{'='*50}")
|
||
print(f"🔔 {title}")
|
||
print(f"{'='*50}")
|
||
print(f"📝 {message}")
|
||
print(f"{'='*50}")
|
||
|
||
if buttons == MB_YESNO:
|
||
print("请选择: [y] 是 [n] 否")
|
||
while True:
|
||
answer = input("> ").strip().lower()
|
||
if answer in ['y', 'yes']:
|
||
return IDYES
|
||
elif answer in ['n', 'no']:
|
||
return IDNO
|
||
else:
|
||
print("请输入 y 或 n")
|
||
elif buttons == MB_YESNOCANCEL:
|
||
print("请选择: [y] 是 [n] 否 [c] 取消")
|
||
while True:
|
||
answer = input("> ").strip().lower()
|
||
if answer in ['y', 'yes']:
|
||
return IDYES
|
||
elif answer in ['n', 'no']:
|
||
return IDNO
|
||
elif answer in ['c', 'cancel']:
|
||
return IDCANCEL
|
||
else:
|
||
print("请输入 y、n 或 c")
|
||
elif buttons == MB_OKCANCEL:
|
||
print("请选择: [o] 确定 [c] 取消")
|
||
while True:
|
||
answer = input("> ").strip().lower()
|
||
if answer in ['o', 'ok', '确定']:
|
||
return IDOK
|
||
elif answer in ['c', 'cancel']:
|
||
return IDCANCEL
|
||
else:
|
||
print("请输入 o 或 c")
|
||
else:
|
||
print("按 Enter 继续...")
|
||
input()
|
||
return IDOK
|
||
|
||
return IDNO
|
||
|
||
def show_input_dialog(title: str, message: str, default: str = "") -> Optional[str]:
|
||
print(f"\n{'='*50}")
|
||
print(f"📝 {title}")
|
||
print(f"{'='*50}")
|
||
print(f"💬 {message}")
|
||
if default:
|
||
print(f"📌 默认值: {default}")
|
||
result = input("请输入: ").strip()
|
||
return result if result else default
|
||
|
||
|
||
def input_text(title: str, message: str, default: str = "") -> Optional[str]:
|
||
"""命令行输入（纯 input，不使用 GUI 弹窗）"""
|
||
return show_input_dialog(title, message, default)
|
||
|
||
# ========== 配置 ==========
|
||
print("=" * 60)
|
||
print("🚀 MCP CMD + SSH 服务器启动 (v3.0.0 - 严格模式)")
|
||
print("=" * 60)
|
||
|
||
def get_initial_workspace():
|
||
workspace_input = input("📂 请输入工作区目录路径: ").strip()
|
||
return workspace_input
|
||
|
||
workspace = get_initial_workspace()
|
||
workspace = os.path.abspath(workspace)
|
||
|
||
if not os.path.exists(workspace):
|
||
print(f"❌ 错误：目录 '{workspace}' 不存在！")
|
||
if is_windows():
|
||
show_confirm_dialog("错误", f"目录 '{workspace}' 不存在！", buttons=MB_OK, icon=MB_ICONERROR)
|
||
exit(1)
|
||
|
||
if not os.path.isdir(workspace):
|
||
print(f"❌ 错误：'{workspace}' 不是目录！")
|
||
if is_windows():
|
||
show_confirm_dialog("错误", f"'{workspace}' 不是目录！", buttons=MB_OK, icon=MB_ICONERROR)
|
||
exit(1)
|
||
|
||
temp_dir = os.path.join(workspace, "temp")
|
||
os.makedirs(temp_dir, exist_ok=True)
|
||
print(f"✅ 临时目录：{temp_dir}")
|
||
print(f"✅ 工作区已设置为：{workspace}")
|
||
print(f"🖥️ 操作系统：{platform.system()}")
|
||
print("=" * 60)
|
||
|
||
# ========== FastAPI 应用 ==========
|
||
app = FastAPI(title="MCP CMD + SSH Server", version="3.0.0")
|
||
|
||
# ========== 全局变量 ==========
|
||
current_workspace = workspace
|
||
current_temp_dir = temp_dir
|
||
|
||
# ========== 🔐 SSH 会话管理 ==========
|
||
class SSHConnectionPool:
|
||
def __init__(self):
|
||
self.sessions: Dict[str, dict] = {}
|
||
self.tokens: Dict[str, str] = {}
|
||
self.session_counter = 0
|
||
|
||
def create_session(self, host: str, username: str, password: str = None,
|
||
key_file: str = None, port: int = 22) -> tuple:
|
||
try:
|
||
client = paramiko.SSHClient()
|
||
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
|
||
|
||
if key_file:
|
||
client.connect(
|
||
host, port=port, username=username,
|
||
key_filename=key_file,
|
||
timeout=15,
|
||
allow_agent=False,
|
||
look_for_keys=False
|
||
)
|
||
else:
|
||
if not password:
|
||
return None, None, False, "❌ 密码或密钥文件必须提供其一"
|
||
client.connect(
|
||
host, port=port, username=username,
|
||
password=password,
|
||
timeout=15,
|
||
allow_agent=False,
|
||
look_for_keys=False
|
||
)
|
||
|
||
channel = client.invoke_shell(
|
||
term='xterm',
|
||
width=200,
|
||
height=50
|
||
)
|
||
|
||
time.sleep(1)
|
||
output = ""
|
||
while channel.recv_ready():
|
||
output += channel.recv(1024).decode('utf-8', errors='ignore')
|
||
|
||
self.session_counter += 1
|
||
session_id = f"ssh_{self.session_counter}_{int(time.time())}"
|
||
token = secrets.token_urlsafe(32)
|
||
|
||
self.sessions[token] = {
|
||
'id': session_id,
|
||
'client': client,
|
||
'channel': channel,
|
||
'host': host,
|
||
'username': username,
|
||
'port': port,
|
||
'created_at': time.time(),
|
||
'last_used': time.time(),
|
||
'buffer': output,
|
||
'is_alive': True
|
||
}
|
||
self.tokens[token] = session_id
|
||
|
||
print(f"✅ SSH会话创建: {session_id} -> {username}@{host}:{port}")
|
||
|
||
return token, session_id, True, f"✅ SSH连接成功！会话ID: {session_id}"
|
||
|
||
except Exception as e:
|
||
print(f"❌ SSH连接失败: {e}")
|
||
return None, None, False, f"❌ SSH连接失败: {str(e)}"
|
||
|
||
def get_session(self, token: str) -> Optional[dict]:
|
||
if token not in self.sessions:
|
||
return None
|
||
|
||
session = self.sessions[token]
|
||
transport = session['client'].get_transport()
|
||
if not session['is_alive'] or transport is None or not transport.is_active():
|
||
self.close_session(token)
|
||
return None
|
||
|
||
session['last_used'] = time.time()
|
||
return session
|
||
|
||
def execute_command(self, token: str, command: str, timeout: int = 30) -> dict:
|
||
session = self.get_session(token)
|
||
if not session:
|
||
return {
|
||
'success': False,
|
||
'output': '',
|
||
'error': '❌ SSH会话已过期或不存在，请重新连接',
|
||
'session_expired': True
|
||
}
|
||
|
||
try:
|
||
channel = session['channel']
|
||
|
||
# 清空缓冲区
|
||
self._flush_buffer(channel)
|
||
|
||
# 发送命令
|
||
channel.send(command + "\n")
|
||
time.sleep(0.2)
|
||
|
||
output = ""
|
||
start_time = time.time()
|
||
|
||
while time.time() - start_time < timeout:
|
||
if channel.recv_ready():
|
||
data = channel.recv(65535).decode('utf-8', errors='ignore')
|
||
output += data
|
||
|
||
if re.search(r'[$#>]', output[-30:]):
|
||
break
|
||
else:
|
||
time.sleep(0.05)
|
||
|
||
session['buffer'] = output
|
||
|
||
return {
|
||
'success': True,
|
||
'output': output,
|
||
'error': '',
|
||
'session_id': session['id']
|
||
}
|
||
|
||
except Exception as e:
|
||
print(f"❌ SSH命令执行失败: {e}")
|
||
return {
|
||
'success': False,
|
||
'output': '',
|
||
'error': f'❌ 命令执行失败: {str(e)}',
|
||
'session_expired': False
|
||
}
|
||
|
||
def _flush_buffer(self, channel):
|
||
while channel.recv_ready():
|
||
channel.recv(1024)
|
||
|
||
def close_session(self, token: str) -> bool:
|
||
if token in self.sessions:
|
||
session = self.sessions[token]
|
||
try:
|
||
session['client'].close()
|
||
except:
|
||
pass
|
||
session['is_alive'] = False
|
||
del self.sessions[token]
|
||
|
||
if token in self.tokens:
|
||
del self.tokens[token]
|
||
|
||
print(f"🗑️ SSH会话已关闭: {token[:16]}...")
|
||
return True
|
||
return False
|
||
|
||
def list_sessions(self) -> list:
|
||
result = []
|
||
for token, session in self.sessions.items():
|
||
transport = session['client'].get_transport()
|
||
if session['is_alive'] and transport is not None and transport.is_active():
|
||
result.append({
|
||
'token': token[:16] + '...',
|
||
'session_id': session['id'],
|
||
'host': session['host'],
|
||
'username': session['username'],
|
||
'created_at': datetime.fromtimestamp(session['created_at']).strftime('%Y-%m-%d %H:%M:%S'),
|
||
'last_used': datetime.fromtimestamp(session['last_used']).strftime('%Y-%m-%d %H:%M:%S')
|
||
})
|
||
return result
|
||
|
||
def cleanup_expired(self, max_age: int = 3600):
|
||
now = time.time()
|
||
to_remove = []
|
||
for token, session in self.sessions.items():
|
||
if now - session['last_used'] > max_age:
|
||
to_remove.append(token)
|
||
|
||
for token in to_remove:
|
||
self.close_session(token)
|
||
|
||
return len(to_remove)
|
||
|
||
# ========== 全局SSH连接池 ==========
|
||
ssh_pool = SSHConnectionPool()
|
||
|
||
# ========== 数据模型 ==========
|
||
class SSHConnectRequest(BaseModel):
|
||
host: str
|
||
username: str
|
||
port: int = 22
|
||
|
||
class SSHExecuteRequest(BaseModel):
|
||
token: str
|
||
command: str
|
||
purpose: str # 必须50-100字说明用途
|
||
timeout: int = 60
|
||
|
||
# ========== 安全函数 ==========
|
||
def ask_user_permission_ui(title: str, message: str, purpose: str = "") -> bool:
|
||
"""命令行确认（不用 GUI 弹窗），输入 y 确认 / n 拒绝"""
|
||
full_message = message
|
||
if purpose:
|
||
full_message += f"\n\n📝 用途说明：\n{purpose}\n"
|
||
|
||
print(f"\n{'='*55}")
|
||
print(f"🔐 {title}")
|
||
print(f"{'='*55}")
|
||
print(full_message)
|
||
print(f"{'='*55}")
|
||
while True:
|
||
answer = input("> 确认请输 y，拒绝输 n [y/n]: ").strip().lower()
|
||
if answer in ('y', 'yes', '是'):
|
||
return True
|
||
elif answer in ('n', 'no', '否'):
|
||
return False
|
||
else:
|
||
print("请输入 y 或 n")
|
||
|
||
def is_path_in_workspace(target_path: str) -> bool:
|
||
global current_workspace
|
||
try:
|
||
target_abs = os.path.abspath(target_path)
|
||
workspace_abs = os.path.abspath(current_workspace)
|
||
return os.path.commonpath([target_abs, workspace_abs]) == workspace_abs
|
||
except ValueError:
|
||
return False
|
||
|
||
def get_safe_cwd(request_cwd: Optional[str]) -> str:
|
||
global current_workspace
|
||
|
||
if not request_cwd:
|
||
return current_workspace
|
||
|
||
request_cwd = os.path.abspath(request_cwd)
|
||
|
||
if is_path_in_workspace(request_cwd):
|
||
return request_cwd
|
||
|
||
if ask_user_permission_ui("目录访问确认", f"命令试图在工作区外的目录执行：\n\n📂 工作区：{current_workspace}\n🎯 目标目录：{request_cwd}\n\n是否允许？"):
|
||
return request_cwd
|
||
else:
|
||
print(f" ❌ 用户拒绝，使用工作区目录：{current_workspace}")
|
||
return current_workspace
|
||
|
||
# ========== 命令执行函数 ==========
|
||
def fix_command_quoting(command: str) -> tuple:
|
||
is_windows_os = platform.system() == "Windows"
|
||
if not is_windows_os:
|
||
return command, False, "非 Windows 系统，无需修复"
|
||
|
||
original = command
|
||
fixed = command
|
||
modifications = []
|
||
|
||
if 'python -c' in command or 'python -c "' in command:
|
||
if '\n' in command and 'python -c' in command:
|
||
match = re.search(r'python\s+-c\s+(["\'])(.*?)\1', command, re.DOTALL)
|
||
if match:
|
||
quote_char = match.group(1)
|
||
code_content = match.group(2)
|
||
|
||
if '\n' in code_content:
|
||
lines = [line.strip() for line in code_content.split('\n') if line.strip()]
|
||
code_lines = []
|
||
for line in lines:
|
||
if line.startswith('#'):
|
||
continue
|
||
if '#' in line and not (line.count("'") >= 2 or line.count('"') >= 2):
|
||
parts = line.split('#')
|
||
if len(parts) > 1 and not (parts[0].count("'") % 2 == 1 or parts[0].count('"') % 2 == 1):
|
||
line = parts[0].strip()
|
||
if line:
|
||
code_lines.append(line)
|
||
|
||
if code_lines:
|
||
single_line = '; '.join(code_lines)
|
||
if quote_char == '"':
|
||
fixed_code = single_line.replace('"', '\\"')
|
||
else:
|
||
fixed_code = single_line
|
||
|
||
fixed = command.replace(match.group(0), f'python -c {quote_char}{fixed_code}{quote_char}')
|
||
modifications.append(f"将多行代码压缩为单行（{len(lines)}行 -> 1行）")
|
||
|
||
if 'python -c' in fixed:
|
||
match = re.search(r'python\s+-c\s+(["\'])(.*?)\1', fixed, re.DOTALL)
|
||
if match:
|
||
outer_quote = match.group(1)
|
||
inner_content = match.group(2)
|
||
|
||
if outer_quote == '"' and '"' in inner_content and '\\"' not in inner_content:
|
||
inner_content = inner_content.replace('"', '\\"')
|
||
fixed = fixed.replace(match.group(0), f'python -c "{inner_content}"')
|
||
modifications.append("转义内部双引号")
|
||
|
||
was_modified = original != fixed
|
||
fix_message = "; ".join(modifications) if modifications else "无需修复"
|
||
|
||
return fixed, was_modified, fix_message
|
||
|
||
def execute_command_safe(command: str, cwd: str, timeout: int = 60) -> dict:
|
||
try:
|
||
is_windows_os = platform.system() == "Windows"
|
||
|
||
fixed_command, was_modified, fix_message = fix_command_quoting(command)
|
||
|
||
if was_modified:
|
||
print(f" 🔧 自动修复命令引号问题")
|
||
print(f" 📝 修复说明：{fix_message}")
|
||
|
||
env = os.environ.copy()
|
||
env['PYTHONIOENCODING'] = 'utf-8'
|
||
env['PYTHONUTF8'] = '1'
|
||
|
||
if is_windows_os:
|
||
shell_command = f'cmd /c "chcp 65001 >nul && set PYTHONIOENCODING=utf-8 && {fixed_command}"'
|
||
else:
|
||
shell_command = fixed_command
|
||
|
||
process = subprocess.Popen(
|
||
shell_command,
|
||
cwd=cwd,
|
||
shell=True,
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.PIPE,
|
||
text=True,
|
||
encoding='utf-8',
|
||
errors='replace',
|
||
bufsize=1,
|
||
universal_newlines=True,
|
||
env=env
|
||
)
|
||
|
||
try:
|
||
stdout, stderr = process.communicate(timeout=timeout)
|
||
return_code = process.returncode
|
||
except subprocess.TimeoutExpired:
|
||
process.kill()
|
||
stdout, stderr = process.communicate()
|
||
return {
|
||
"success": False,
|
||
"output": stdout or "",
|
||
"error": f"⏰ 命令执行超时（{timeout}秒）",
|
||
"return_code": -1,
|
||
"cwd": cwd,
|
||
"command": command
|
||
}
|
||
|
||
output = stdout or ""
|
||
error = stderr or ""
|
||
success = return_code == 0
|
||
|
||
if not success and "Argument expected for the -c option" in error:
|
||
return {
|
||
"success": False,
|
||
"output": output,
|
||
"error": error,
|
||
"return_code": return_code,
|
||
"cwd": cwd,
|
||
"command": command,
|
||
"suggestion": "💡 建议使用 python_runner 工具"
|
||
}
|
||
|
||
if not output and error:
|
||
if "warning" not in error.lower() and "error" not in error.lower():
|
||
output = error
|
||
error = ""
|
||
|
||
return {
|
||
"success": success,
|
||
"output": output,
|
||
"error": error,
|
||
"return_code": return_code,
|
||
"cwd": cwd,
|
||
"command": command
|
||
}
|
||
|
||
except Exception as e:
|
||
return {
|
||
"success": False,
|
||
"output": "",
|
||
"error": f"❌ 执行失败：{str(e)}",
|
||
"return_code": -1,
|
||
"cwd": cwd,
|
||
"command": command
|
||
}
|
||
|
||
def create_and_run_python(code: str, timeout: int = 60, cwd: Optional[str] = None) -> dict:
|
||
global current_temp_dir, current_workspace
|
||
|
||
try:
|
||
file_id = uuid.uuid4().hex[:8]
|
||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
filename = f"temp_{timestamp}_{file_id}.py"
|
||
filepath = os.path.join(current_temp_dir, filename)
|
||
|
||
with open(filepath, 'w', encoding='utf-8') as f:
|
||
f.write(code)
|
||
|
||
exec_cwd = cwd or current_workspace
|
||
python_cmd = f'python "{filepath}"'
|
||
|
||
result = execute_command_safe(python_cmd, exec_cwd, timeout)
|
||
result['file_created'] = filepath
|
||
result['file_name'] = filename
|
||
|
||
return result
|
||
|
||
except Exception as e:
|
||
return {
|
||
"success": False,
|
||
"output": "",
|
||
"error": f"❌ 创建或执行失败：{str(e)}",
|
||
"return_code": -1,
|
||
"cwd": cwd or current_workspace,
|
||
"command": code[:100] + "..."
|
||
}
|
||
|
||
def switch_workspace(new_workspace: str) -> dict:
|
||
global current_workspace, current_temp_dir
|
||
|
||
new_workspace = os.path.abspath(new_workspace)
|
||
|
||
if not os.path.exists(new_workspace):
|
||
return {"success": False, "error": f"目录 '{new_workspace}' 不存在！"}
|
||
|
||
if not os.path.isdir(new_workspace):
|
||
return {"success": False, "error": f"'{new_workspace}' 不是目录！"}
|
||
|
||
old_workspace = current_workspace
|
||
|
||
message = f"确认切换工作区？\n\n📂 当前工作区：{old_workspace}\n📂 新工作区：{new_workspace}"
|
||
|
||
result = show_confirm_dialog("🔄 切换工作区", message, buttons=MB_YESNO, icon=MB_ICONQUESTION)
|
||
|
||
if result != IDYES:
|
||
return {"success": False, "error": "用户取消切换", "current_workspace": current_workspace}
|
||
|
||
current_workspace = new_workspace
|
||
current_temp_dir = os.path.join(current_workspace, "temp")
|
||
os.makedirs(current_temp_dir, exist_ok=True)
|
||
|
||
return {
|
||
"success": True,
|
||
"old_workspace": old_workspace,
|
||
"new_workspace": current_workspace,
|
||
"temp_dir": current_temp_dir,
|
||
"message": f"✅ 成功切换到 {current_workspace}"
|
||
}
|
||
|
||
# ========== MCP 核心端点 ==========
|
||
@app.post("/mcp")
|
||
async def mcp_handler(request: Request):
|
||
global current_workspace, current_temp_dir
|
||
|
||
try:
|
||
body = await request.json()
|
||
method = body.get("method")
|
||
params = body.get("params", {})
|
||
request_id = body.get("id")
|
||
|
||
if method == "initialize":
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"result": {
|
||
"protocolVersion": "1.0.0",
|
||
"serverInfo": {
|
||
"name": "mcp-cmd-ssh-server",
|
||
"version": "3.0.0"
|
||
},
|
||
"capabilities": {"tools": {}}
|
||
}
|
||
}
|
||
|
||
elif method == "tools/list":
|
||
tools_response = {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"result": {
|
||
"tools": [
|
||
{
|
||
"name": "command_executor",
|
||
"description": f"执行系统命令。默认在工作区 '{current_workspace}' 执行。",
|
||
"inputSchema": {
|
||
"type": "object",
|
||
"properties": {
|
||
"command": {"type": "string", "description": "要执行的命令"},
|
||
"cwd": {"type": "string", "description": f"可选，执行目录。默认工作区 '{current_workspace}'"},
|
||
"timeout": {"type": "integer", "description": "可选，超时时间（秒），默认 60", "default": 60}
|
||
},
|
||
"required": ["command"]
|
||
}
|
||
},
|
||
{
|
||
"name": "python_runner",
|
||
"description": "创建临时 Python 文件并执行。",
|
||
"inputSchema": {
|
||
"type": "object",
|
||
"properties": {
|
||
"code": {"type": "string", "description": "Python 代码内容"},
|
||
"cwd": {"type": "string", "description": f"可选，执行目录。默认工作区 '{current_workspace}'"},
|
||
"timeout": {"type": "integer", "description": "可选，超时时间（秒），默认 60", "default": 60}
|
||
},
|
||
"required": ["code"]
|
||
}
|
||
},
|
||
{
|
||
"name": "ssh_connect",
|
||
"description": "🔐 发起SSH连接请求。用户需手动输入密码，成功返回token。",
|
||
"inputSchema": {
|
||
"type": "object",
|
||
"properties": {
|
||
"host": {"type": "string", "description": "SSH服务器地址"},
|
||
"username": {"type": "string", "description": "SSH用户名"},
|
||
"port": {"type": "integer", "description": "SSH端口，默认22", "default": 22}
|
||
},
|
||
"required": ["host", "username"]
|
||
}
|
||
},
|
||
{
|
||
"name": "ssh_execute",
|
||
"description": "🔐 在SSH会话中执行命令。⚠️ 必须提供50-100字的用途说明，每次执行都需要用户命令行确认。",
|
||
"inputSchema": {
|
||
"type": "object",
|
||
"properties": {
|
||
"token": {"type": "string", "description": "SSH连接返回的token"},
|
||
"command": {"type": "string", "description": "要执行的命令"},
|
||
"purpose": {"type": "string", "description": "⚠️ 用途说明（必须50-100字），详细说明为什么执行此命令、预期效果、涉及的操作"},
|
||
"timeout": {"type": "integer", "description": "可选，超时时间（秒），默认60", "default": 60}
|
||
},
|
||
"required": ["token", "command", "purpose"]
|
||
}
|
||
},
|
||
{
|
||
"name": "ssh_close",
|
||
"description": "🔐 关闭SSH会话",
|
||
"inputSchema": {
|
||
"type": "object",
|
||
"properties": {
|
||
"token": {"type": "string", "description": "SSH连接返回的token"}
|
||
},
|
||
"required": ["token"]
|
||
}
|
||
},
|
||
{
|
||
"name": "ssh_list_sessions",
|
||
"description": "🔐 列出所有活跃的SSH会话",
|
||
"inputSchema": {"type": "object", "properties": {}}
|
||
},
|
||
{
|
||
"name": "file_reader",
|
||
"description": "读取工作区内文件的内容",
|
||
"inputSchema": {
|
||
"type": "object",
|
||
"properties": {
|
||
"filepath": {"type": "string", "description": "文件路径"},
|
||
"encoding": {"type": "string", "description": "文件编码，默认 utf-8", "default": "utf-8"}
|
||
},
|
||
"required": ["filepath"]
|
||
}
|
||
},
|
||
{
|
||
"name": "file_writer",
|
||
"description": "在工作区内创建或修改文件",
|
||
"inputSchema": {
|
||
"type": "object",
|
||
"properties": {
|
||
"filepath": {"type": "string", "description": "文件路径"},
|
||
"content": {"type": "string", "description": "要写入的内容"},
|
||
"mode": {"type": "string", "description": "写入模式：'w' 覆盖，'a' 追加", "enum": ["w", "a"], "default": "w"}
|
||
},
|
||
"required": ["filepath", "content"]
|
||
}
|
||
},
|
||
{
|
||
"name": "directory_lister",
|
||
"description": "列出指定目录的内容",
|
||
"inputSchema": {
|
||
"type": "object",
|
||
"properties": {
|
||
"path": {"type": "string", "description": f"目录路径。默认工作区 '{current_workspace}'"},
|
||
"show_hidden": {"type": "boolean", "description": "是否显示隐藏文件，默认 false", "default": False}
|
||
}
|
||
}
|
||
},
|
||
{
|
||
"name": "file_searcher",
|
||
"description": "在工作区内搜索文件",
|
||
"inputSchema": {
|
||
"type": "object",
|
||
"properties": {
|
||
"pattern": {"type": "string", "description": "搜索模式（如 '*.txt'）"},
|
||
"search_root": {"type": "string", "description": f"搜索根目录。默认工作区 '{current_workspace}'"},
|
||
"recursive": {"type": "boolean", "description": "是否递归搜索，默认 true", "default": True}
|
||
},
|
||
"required": ["pattern"]
|
||
}
|
||
},
|
||
{
|
||
"name": "git_operations",
|
||
"description": "在工作区内执行 Git 操作",
|
||
"inputSchema": {
|
||
"type": "object",
|
||
"properties": {
|
||
"git_command": {"type": "string", "description": "Git 命令"},
|
||
"repo_path": {"type": "string", "description": f"Git 仓库路径。默认工作区 '{current_workspace}'"}
|
||
},
|
||
"required": ["git_command"]
|
||
}
|
||
},
|
||
{
|
||
"name": "switch_workspace",
|
||
"description": f"🔄 切换工作区。当前：'{current_workspace}'",
|
||
"inputSchema": {
|
||
"type": "object",
|
||
"properties": {
|
||
"new_workspace": {"type": "string", "description": "新的工作区目录路径"}
|
||
},
|
||
"required": ["new_workspace"]
|
||
}
|
||
},
|
||
{
|
||
"name": "get_current_workspace",
|
||
"description": "📂 获取当前工作区信息",
|
||
"inputSchema": {"type": "object", "properties": {}}
|
||
}
|
||
]
|
||
}
|
||
}
|
||
|
||
print(f"\n📤 返回工具列表，共 {len(tools_response['result']['tools'])} 个工具")
|
||
return tools_response
|
||
|
||
elif method == "tools/call":
|
||
tool_name = params.get("name")
|
||
arguments = params.get("arguments", {})
|
||
|
||
# ========== command_executor ==========
|
||
if tool_name == "command_executor":
|
||
command = arguments.get("command")
|
||
cwd = arguments.get("cwd")
|
||
timeout = arguments.get("timeout", 60)
|
||
|
||
if not command:
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"error": {"code": -32602, "message": "❌ 缺少 command 参数"}
|
||
}
|
||
|
||
safe_cwd = get_safe_cwd(cwd)
|
||
result = execute_command_safe(command, safe_cwd, timeout)
|
||
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"result": {
|
||
"content": [{"type": "text", "text": json.dumps(result, ensure_ascii=False, indent=2)}]
|
||
}
|
||
}
|
||
|
||
# ========== python_runner ==========
|
||
elif tool_name == "python_runner":
|
||
code = arguments.get("code")
|
||
cwd = arguments.get("cwd")
|
||
timeout = arguments.get("timeout", 60)
|
||
|
||
if not code:
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"error": {"code": -32602, "message": "❌ 缺少 code 参数"}
|
||
}
|
||
|
||
safe_cwd = get_safe_cwd(cwd)
|
||
result = create_and_run_python(code, timeout, safe_cwd)
|
||
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"result": {
|
||
"content": [{"type": "text", "text": json.dumps(result, ensure_ascii=False, indent=2)}]
|
||
}
|
||
}
|
||
|
||
# ========== 🔐 ssh_connect ==========
|
||
elif tool_name == "ssh_connect":
|
||
host = arguments.get("host")
|
||
username = arguments.get("username")
|
||
port = arguments.get("port", 22)
|
||
|
||
if not host or not username:
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"error": {"code": -32602, "message": "❌ 缺少 host 或 username 参数"}
|
||
}
|
||
|
||
# 命令行询问密码（使用 input，不用 GUI 弹窗）
|
||
print(f"\n🔐 请输入 {username}@{host}:{port} 的 SSH 密码: ", end="", flush=True)
|
||
password = input().strip()
|
||
|
||
if not password:
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"result": {
|
||
"content": [{
|
||
"type": "text",
|
||
"text": json.dumps({
|
||
"success": False,
|
||
"message": "❌ 用户未输入密码，连接取消"
|
||
}, ensure_ascii=False, indent=2)
|
||
}]
|
||
}
|
||
}
|
||
|
||
token, session_id, success, message = ssh_pool.create_session(
|
||
host, username, password, port=port
|
||
)
|
||
|
||
result = {
|
||
"success": success,
|
||
"message": message,
|
||
"token": token if success else None,
|
||
"session_id": session_id if success else None,
|
||
"host": host,
|
||
"username": username,
|
||
"port": port
|
||
}
|
||
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"result": {
|
||
"content": [{"type": "text", "text": json.dumps(result, ensure_ascii=False, indent=2)}]
|
||
}
|
||
}
|
||
|
||
# ========== 🔐 ssh_execute（严格模式） ==========
|
||
elif tool_name == "ssh_execute":
|
||
token = arguments.get("token")
|
||
command = arguments.get("command")
|
||
purpose = arguments.get("purpose", "")
|
||
timeout = arguments.get("timeout", 60)
|
||
|
||
if not token or not command:
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"error": {"code": -32602, "message": "❌ 缺少 token 或 command 参数"}
|
||
}
|
||
|
||
# ========== 严格检查：用途必须50-100字 ==========
|
||
purpose_len = len(purpose)
|
||
if purpose_len < 50:
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"result": {
|
||
"content": [{
|
||
"type": "text",
|
||
"text": json.dumps({
|
||
"success": False,
|
||
"error": f"❌ 用途说明字数不足！当前 {purpose_len} 字，必须 50-100 字。",
|
||
"requirement": "请提供至少50字的详细用途说明，包括：为什么要执行此命令、预期达到什么效果、涉及哪些操作和风险。",
|
||
"current_purpose": purpose,
|
||
"purpose_length": purpose_len,
|
||
"minimum_required": 50,
|
||
"maximum_required": 100
|
||
}, ensure_ascii=False, indent=2)
|
||
}]
|
||
}
|
||
}
|
||
|
||
if purpose_len > 100:
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"result": {
|
||
"content": [{
|
||
"type": "text",
|
||
"text": json.dumps({
|
||
"success": False,
|
||
"error": f"❌ 用途说明超长！当前 {purpose_len} 字，必须 50-100 字。",
|
||
"requirement": "请精简用途说明到100字以内，保持清晰简洁。",
|
||
"current_purpose": purpose,
|
||
"purpose_length": purpose_len,
|
||
"minimum_required": 50,
|
||
"maximum_required": 100
|
||
}, ensure_ascii=False, indent=2)
|
||
}]
|
||
}
|
||
}
|
||
|
||
# 获取会话信息
|
||
session = ssh_pool.get_session(token)
|
||
if not session:
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"result": {
|
||
"content": [{
|
||
"type": "text",
|
||
"text": json.dumps({
|
||
"success": False,
|
||
"error": "❌ SSH会话已过期或不存在，请重新连接",
|
||
"token": token[:16] + "..."
|
||
}, ensure_ascii=False, indent=2)
|
||
}]
|
||
}
|
||
}
|
||
|
||
# ========== 🔐 用户确认（命令行 input） ==========
|
||
confirm_message = (
|
||
f"SSH会话：{session['username']}@{session['host']}:{session['port']}\n"
|
||
f"会话ID：{session['id']}\n\n"
|
||
f"📝 用途说明（{purpose_len}字）：\n{purpose}\n\n"
|
||
f"💻 命令：\n{command[:300]}{'...' if len(command) > 300 else ''}\n\n"
|
||
f"⚠️ 请确认此操作是您期望的，且了解可能的风险。"
|
||
)
|
||
|
||
if not ask_user_permission_ui("SSH命令执行确认", confirm_message):
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"result": {
|
||
"content": [{
|
||
"type": "text",
|
||
"text": json.dumps({
|
||
"success": False,
|
||
"error": "❌ 用户拒绝执行SSH命令",
|
||
"command": command,
|
||
"purpose": purpose,
|
||
"purpose_length": purpose_len
|
||
}, ensure_ascii=False, indent=2)
|
||
}]
|
||
}
|
||
}
|
||
|
||
# 执行命令
|
||
result = ssh_pool.execute_command(token, command, timeout)
|
||
result['purpose'] = purpose
|
||
result['purpose_length'] = purpose_len
|
||
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"result": {
|
||
"content": [{"type": "text", "text": json.dumps(result, ensure_ascii=False, indent=2)}]
|
||
}
|
||
}
|
||
|
||
# ========== 🔐 ssh_close ==========
|
||
elif tool_name == "ssh_close":
|
||
token = arguments.get("token")
|
||
|
||
if not token:
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"error": {"code": -32602, "message": "❌ 缺少 token 参数"}
|
||
}
|
||
|
||
success = ssh_pool.close_session(token)
|
||
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"result": {
|
||
"content": [{
|
||
"type": "text",
|
||
"text": json.dumps({
|
||
"success": success,
|
||
"message": "✅ SSH会话已关闭" if success else "❌ 会话不存在或已关闭",
|
||
"token": token[:16] + "..."
|
||
}, ensure_ascii=False, indent=2)
|
||
}]
|
||
}
|
||
}
|
||
|
||
# ========== 🔐 ssh_list_sessions ==========
|
||
elif tool_name == "ssh_list_sessions":
|
||
sessions = ssh_pool.list_sessions()
|
||
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"result": {
|
||
"content": [{
|
||
"type": "text",
|
||
"text": json.dumps({
|
||
"success": True,
|
||
"total": len(sessions),
|
||
"sessions": sessions
|
||
}, ensure_ascii=False, indent=2)
|
||
}]
|
||
}
|
||
}
|
||
|
||
# ========== file_reader ==========
|
||
elif tool_name == "file_reader":
|
||
filepath = arguments.get("filepath")
|
||
encoding = arguments.get("encoding", "utf-8")
|
||
|
||
if not filepath:
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"error": {"code": -32602, "message": "❌ 缺少 filepath 参数"}
|
||
}
|
||
|
||
if os.path.isabs(filepath):
|
||
target_path = filepath
|
||
else:
|
||
target_path = os.path.join(current_workspace, filepath)
|
||
|
||
target_path = os.path.abspath(target_path)
|
||
|
||
if not is_path_in_workspace(target_path):
|
||
if not ask_user_permission_ui("文件读取确认", f"读取工作区外的文件：\n📂 文件：{target_path}\n\n是否允许？"):
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"result": {
|
||
"content": [{
|
||
"type": "text",
|
||
"text": json.dumps({
|
||
"success": False,
|
||
"error": f"❌ 用户拒绝访问：{target_path}"
|
||
}, ensure_ascii=False, indent=2)
|
||
}]
|
||
}
|
||
}
|
||
|
||
try:
|
||
with open(target_path, 'r', encoding=encoding) as f:
|
||
content = f.read()
|
||
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"result": {
|
||
"content": [{
|
||
"type": "text",
|
||
"text": json.dumps({
|
||
"success": True,
|
||
"content": content,
|
||
"filepath": target_path,
|
||
"size": len(content)
|
||
}, ensure_ascii=False, indent=2)
|
||
}]
|
||
}
|
||
}
|
||
except Exception as e:
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"result": {
|
||
"content": [{
|
||
"type": "text",
|
||
"text": json.dumps({
|
||
"success": False,
|
||
"error": f"❌ 读取失败：{str(e)}",
|
||
"filepath": target_path
|
||
}, ensure_ascii=False, indent=2)
|
||
}]
|
||
}
|
||
}
|
||
|
||
# ========== file_writer ==========
|
||
elif tool_name == "file_writer":
|
||
filepath = arguments.get("filepath")
|
||
content = arguments.get("content")
|
||
mode = arguments.get("mode", "w")
|
||
|
||
if not filepath or content is None:
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"error": {"code": -32602, "message": "❌ 缺少 filepath 或 content 参数"}
|
||
}
|
||
|
||
if os.path.isabs(filepath):
|
||
target_path = filepath
|
||
else:
|
||
target_path = os.path.join(current_workspace, filepath)
|
||
|
||
target_path = os.path.abspath(target_path)
|
||
|
||
if not is_path_in_workspace(target_path):
|
||
if not ask_user_permission_ui("文件写入确认", f"写入工作区外的文件：\n📂 文件：{target_path}\n\n是否允许？"):
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"result": {
|
||
"content": [{
|
||
"type": "text",
|
||
"text": json.dumps({
|
||
"success": False,
|
||
"error": f"❌ 用户拒绝写入：{target_path}"
|
||
}, ensure_ascii=False, indent=2)
|
||
}]
|
||
}
|
||
}
|
||
|
||
try:
|
||
os.makedirs(os.path.dirname(target_path), exist_ok=True)
|
||
with open(target_path, mode, encoding='utf-8') as f:
|
||
f.write(content)
|
||
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"result": {
|
||
"content": [{
|
||
"type": "text",
|
||
"text": json.dumps({
|
||
"success": True,
|
||
"filepath": target_path,
|
||
"mode": mode,
|
||
"size": len(content),
|
||
"message": f"✅ 文件已{'追加' if mode == 'a' else '写入'}"
|
||
}, ensure_ascii=False, indent=2)
|
||
}]
|
||
}
|
||
}
|
||
except Exception as e:
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"result": {
|
||
"content": [{
|
||
"type": "text",
|
||
"text": json.dumps({
|
||
"success": False,
|
||
"error": f"❌ 写入失败：{str(e)}",
|
||
"filepath": target_path
|
||
}, ensure_ascii=False, indent=2)
|
||
}]
|
||
}
|
||
}
|
||
|
||
# ========== directory_lister ==========
|
||
elif tool_name == "directory_lister":
|
||
path = arguments.get("path", current_workspace)
|
||
show_hidden = arguments.get("show_hidden", False)
|
||
|
||
if os.path.isabs(path):
|
||
target_path = path
|
||
else:
|
||
target_path = os.path.join(current_workspace, path)
|
||
|
||
target_path = os.path.abspath(target_path)
|
||
|
||
if not is_path_in_workspace(target_path):
|
||
if not ask_user_permission_ui("目录列表确认", f"列出工作区外的目录：\n📂 目录：{target_path}\n\n是否允许？"):
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"result": {
|
||
"content": [{
|
||
"type": "text",
|
||
"text": json.dumps({
|
||
"success": False,
|
||
"error": f"❌ 用户拒绝访问：{target_path}"
|
||
}, ensure_ascii=False, indent=2)
|
||
}]
|
||
}
|
||
}
|
||
|
||
try:
|
||
items = []
|
||
for item in os.listdir(target_path):
|
||
if not show_hidden and item.startswith('.'):
|
||
continue
|
||
|
||
full_path = os.path.join(target_path, item)
|
||
is_dir = os.path.isdir(full_path)
|
||
size = os.path.getsize(full_path) if not is_dir else 0
|
||
mtime = datetime.fromtimestamp(os.path.getmtime(full_path)).strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
items.append({
|
||
"name": item,
|
||
"type": "目录" if is_dir else "文件",
|
||
"size": size,
|
||
"modified": mtime,
|
||
"path": full_path
|
||
})
|
||
|
||
items.sort(key=lambda x: (x['type'] != '目录', x['name'].lower()))
|
||
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"result": {
|
||
"content": [{
|
||
"type": "text",
|
||
"text": json.dumps({
|
||
"success": True,
|
||
"path": target_path,
|
||
"items": items,
|
||
"count": len(items)
|
||
}, ensure_ascii=False, indent=2)
|
||
}]
|
||
}
|
||
}
|
||
except Exception as e:
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"result": {
|
||
"content": [{
|
||
"type": "text",
|
||
"text": json.dumps({
|
||
"success": False,
|
||
"error": f"❌ 列出目录失败：{str(e)}",
|
||
"path": target_path
|
||
}, ensure_ascii=False, indent=2)
|
||
}]
|
||
}
|
||
}
|
||
|
||
# ========== file_searcher ==========
|
||
elif tool_name == "file_searcher":
|
||
pattern = arguments.get("pattern")
|
||
search_root = arguments.get("search_root", current_workspace)
|
||
recursive = arguments.get("recursive", True)
|
||
|
||
if not pattern:
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"error": {"code": -32602, "message": "❌ 缺少 pattern 参数"}
|
||
}
|
||
|
||
if os.path.isabs(search_root):
|
||
root_path = search_root
|
||
else:
|
||
root_path = os.path.join(current_workspace, search_root)
|
||
|
||
root_path = os.path.abspath(root_path)
|
||
|
||
if not is_path_in_workspace(root_path):
|
||
if not ask_user_permission_ui("文件搜索确认", f"在工作区外搜索文件：\n📂 目录：{root_path}\n\n是否允许？"):
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"result": {
|
||
"content": [{
|
||
"type": "text",
|
||
"text": json.dumps({
|
||
"success": False,
|
||
"error": f"❌ 用户拒绝搜索：{root_path}"
|
||
}, ensure_ascii=False, indent=2)
|
||
}]
|
||
}
|
||
}
|
||
|
||
try:
|
||
matches = []
|
||
search_pattern = pattern.replace('*', '.*').replace('?', '.')
|
||
regex = re.compile(search_pattern, re.IGNORECASE)
|
||
|
||
if recursive:
|
||
for dirpath, dirnames, filenames in os.walk(root_path):
|
||
dirnames[:] = [d for d in dirnames if not d.startswith('.')]
|
||
for filename in filenames:
|
||
if regex.match(filename):
|
||
full_path = os.path.join(dirpath, filename)
|
||
matches.append(full_path)
|
||
else:
|
||
for item in os.listdir(root_path):
|
||
if regex.match(item):
|
||
full_path = os.path.join(root_path, item)
|
||
if os.path.isfile(full_path):
|
||
matches.append(full_path)
|
||
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"result": {
|
||
"content": [{
|
||
"type": "text",
|
||
"text": json.dumps({
|
||
"success": True,
|
||
"pattern": pattern,
|
||
"search_root": root_path,
|
||
"recursive": recursive,
|
||
"matches": matches,
|
||
"count": len(matches)
|
||
}, ensure_ascii=False, indent=2)
|
||
}]
|
||
}
|
||
}
|
||
except Exception as e:
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"result": {
|
||
"content": [{
|
||
"type": "text",
|
||
"text": json.dumps({
|
||
"success": False,
|
||
"error": f"❌ 搜索失败：{str(e)}",
|
||
"pattern": pattern
|
||
}, ensure_ascii=False, indent=2)
|
||
}]
|
||
}
|
||
}
|
||
|
||
# ========== git_operations ==========
|
||
elif tool_name == "git_operations":
|
||
git_command = arguments.get("git_command")
|
||
repo_path = arguments.get("repo_path", current_workspace)
|
||
|
||
if not git_command:
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"error": {"code": -32602, "message": "❌ 缺少 git_command 参数"}
|
||
}
|
||
|
||
if os.path.isabs(repo_path):
|
||
target_path = repo_path
|
||
else:
|
||
target_path = os.path.join(current_workspace, repo_path)
|
||
|
||
target_path = os.path.abspath(target_path)
|
||
|
||
if not is_path_in_workspace(target_path):
|
||
if not ask_user_permission_ui("Git操作确认", f"在工作区外执行Git操作：\n📂 仓库：{target_path}\n\n是否允许？"):
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"result": {
|
||
"content": [{
|
||
"type": "text",
|
||
"text": json.dumps({
|
||
"success": False,
|
||
"error": f"❌ 用户拒绝 Git 操作：{target_path}"
|
||
}, ensure_ascii=False, indent=2)
|
||
}]
|
||
}
|
||
}
|
||
|
||
full_command = f"git {git_command}"
|
||
result = execute_command_safe(full_command, target_path)
|
||
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"result": {
|
||
"content": [{"type": "text", "text": json.dumps(result, ensure_ascii=False, indent=2)}]
|
||
}
|
||
}
|
||
|
||
# ========== switch_workspace ==========
|
||
elif tool_name == "switch_workspace":
|
||
new_workspace = arguments.get("new_workspace")
|
||
|
||
if not new_workspace:
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"error": {"code": -32602, "message": "❌ 缺少 new_workspace 参数"}
|
||
}
|
||
|
||
result = switch_workspace(new_workspace)
|
||
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"result": {
|
||
"content": [{"type": "text", "text": json.dumps(result, ensure_ascii=False, indent=2)}]
|
||
}
|
||
}
|
||
|
||
# ========== get_current_workspace ==========
|
||
elif tool_name == "get_current_workspace":
|
||
result = {
|
||
"workspace": current_workspace,
|
||
"temp_dir": current_temp_dir,
|
||
"os": platform.system(),
|
||
"version": "3.0.0"
|
||
}
|
||
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"result": {
|
||
"content": [{"type": "text", "text": json.dumps(result, ensure_ascii=False, indent=2)}]
|
||
}
|
||
}
|
||
|
||
else:
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"error": {"code": -32601, "message": f"❌ 未知工具：{tool_name}"}
|
||
}
|
||
|
||
else:
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": request_id,
|
||
"error": {"code": -32601, "message": f"❌ 未知方法：{method}"}
|
||
}
|
||
|
||
except Exception as e:
|
||
print(f"\n❌ 服务器错误：{str(e)}")
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"id": body.get("id", None),
|
||
"error": {"code": -32603, "message": f"❌ 内部服务器错误：{str(e)}"}
|
||
}
|
||
|
||
# ========== 辅助API ==========
|
||
@app.post("/switch_workspace")
|
||
async def switch_workspace_api(request: Request):
|
||
global current_workspace, current_temp_dir
|
||
|
||
try:
|
||
body = await request.json()
|
||
new_workspace = body.get("workspace", "").strip()
|
||
|
||
if not new_workspace:
|
||
return JSONResponse({"success": False, "error": "请提供 workspace 参数"})
|
||
|
||
result = switch_workspace(new_workspace)
|
||
return JSONResponse(result)
|
||
|
||
except Exception as e:
|
||
return JSONResponse({"success": False, "error": f"切换失败：{str(e)}"})
|
||
|
||
@app.get("/current_workspace")
|
||
async def get_current_workspace_api():
|
||
global current_workspace, current_temp_dir
|
||
return {
|
||
"workspace": current_workspace,
|
||
"temp_dir": current_temp_dir,
|
||
"os": platform.system(),
|
||
"version": "3.0.0"
|
||
}
|
||
|
||
@app.get("/ssh/sessions")
|
||
async def list_ssh_sessions():
|
||
"""列出所有SSH会话（管理用）"""
|
||
sessions = ssh_pool.list_sessions()
|
||
return {
|
||
"success": True,
|
||
"total": len(sessions),
|
||
"sessions": sessions
|
||
}
|
||
|
||
@app.post("/ssh/cleanup")
|
||
async def cleanup_ssh_sessions():
|
||
"""清理过期SSH会话"""
|
||
count = ssh_pool.cleanup_expired()
|
||
return {
|
||
"success": True,
|
||
"cleaned": count,
|
||
"message": f"✅ 清理了 {count} 个过期会话"
|
||
}
|
||
|
||
@app.get("/health")
|
||
async def health():
|
||
global current_workspace, current_temp_dir
|
||
sessions = ssh_pool.list_sessions()
|
||
return {
|
||
"status": "ok",
|
||
"workspace": current_workspace,
|
||
"temp_dir": current_temp_dir,
|
||
"os": platform.system(),
|
||
"version": "3.0.0",
|
||
"ssh_sessions": len(sessions),
|
||
"features": [
|
||
"SSH交互支持",
|
||
"严格50-100字用途说明",
|
||
"每次命令命令行确认",
|
||
"会话Token管理",
|
||
"自动清理过期会话"
|
||
]
|
||
}
|
||
|
||
@app.get("/")
|
||
async def root():
|
||
global current_workspace, current_temp_dir
|
||
return {
|
||
"message": "🚀 MCP CMD + SSH Server v3.0.0 (严格模式)",
|
||
"workspace": current_workspace,
|
||
"temp_dir": current_temp_dir,
|
||
"endpoints": {
|
||
"mcp": "POST /mcp",
|
||
"health": "GET /health",
|
||
"current_workspace": "GET /current_workspace",
|
||
"switch_workspace": "POST /switch_workspace",
|
||
"ssh_sessions": "GET /ssh/sessions",
|
||
"ssh_cleanup": "POST /ssh/cleanup"
|
||
},
|
||
"ssh_features": {
|
||
"connect": "用户手动输入密码，返回token",
|
||
"execute": "必须50-100字用途说明 + 命令行确认",
|
||
"close": "关闭会话",
|
||
"list": "查看所有会话"
|
||
},
|
||
"strict_rules": {
|
||
"purpose_min_length": 50,
|
||
"purpose_max_length": 100,
|
||
"confirmation_required": True,
|
||
"per_command_confirmation": True
|
||
}
|
||
}
|
||
|
||
# ========== 启动服务器 ==========
|
||
if __name__ == "__main__":
|
||
import uvicorn
|
||
print(f"\n🔥 MCP + SSH 服务器启动中...")
|
||
print(f"📍 主端点：http://localhost:3000/mcp")
|
||
print(f"📂 工作区：{workspace}")
|
||
print(f"📁 临时目录：{temp_dir}")
|
||
print(f"\n🔐 SSH功能：")
|
||
print(f" - ssh_connect: 发起连接，用户手动输入密码")
|
||
print(f" - ssh_execute: 执行命令（⚠️ 必须50-100字用途说明 + 命令行确认）")
|
||
print(f" - ssh_close: 关闭会话")
|
||
print(f" - ssh_list_sessions: 列出会话")
|
||
print(f"\n📋 严格规则：")
|
||
print(f" - 用途说明：50-100字（不足50字拒绝，超过100字拒绝）")
|
||
print(f" - 每次执行必须命令行确认")
|
||
print(f" - 每个命令单独审批")
|
||
print(f"\n💡 按 Ctrl+C 停止服务器")
|
||
print("=" * 60)
|
||
uvicorn.run(app, host="0.0.0.0", port=3000, log_level="info")
|