提交 208d5e63 authored 作者: 陈泽健's avatar 陈泽健

fix(smart-locate): Claude服务支持Docker容器内通过SSH调用宿主机CLI

- 新增3种调用方式(按优先级):本地CLI > SSH宿主机 > 容器内CLI
- 新增_call_claude_via_ssh方法:通过SSH调用宿主机Claude CLI
- 新增_call_claude_via_paramiko方法:paramiko回退方案
- 支持环境变量配置SSH连接参数(CLAUDE_SSH_HOST/PORT/USER/PASS)
- 已部署到192.168.5.60验证通过
Co-Authored-By: 's avatarClaude <noreply@anthropic.com>
上级 85b7cb33
......@@ -13,6 +13,7 @@ import logging
import subprocess
import json
import re
import os
from typing import List, Dict, Any, Tuple, Optional
from app.config import settings
......@@ -177,6 +178,11 @@ class ClaudeService:
"""
调用 Claude CLI
支持三种调用方式(按优先级):
1. 本地 Claude CLI(Windows/Linux 本地环境)
2. SSH 调用宿主机 Claude CLI(Docker 容器内环境)
3. 容器内 Claude CLI(如果已安装)
Args:
prompt: Prompt 文本
......@@ -189,17 +195,16 @@ class ClaudeService:
import platform
import os
try:
# 检测操作系统
is_windows = platform.system() == 'Windows'
# 设置环境变量确保 UTF-8 编码
env = os.environ.copy()
env['PYTHONIOENCODING'] = 'utf-8'
env['LANG'] = 'en_US.UTF-8'
is_windows = platform.system() == 'Windows'
# ========== 方式1: 本地 Claude CLI ==========
try:
if is_windows:
# Windows: 使用 shell=True 并设置编码
result = subprocess.run(
f'claude --print',
input=prompt,
......@@ -209,10 +214,9 @@ class ClaudeService:
shell=True,
env=env,
encoding='utf-8',
errors='replace' # 替换无法解码的字符
errors='replace'
)
else:
# Linux/Mac: 直接调用
result = subprocess.run(
['claude', '--print'],
input=prompt,
......@@ -224,26 +228,168 @@ class ClaudeService:
errors='replace'
)
# 检查结果
stdout = result.stdout.strip() if result.stdout else ""
stderr = result.stderr.strip() if result.stderr else ""
if result.returncode == 0 and result.stdout and result.stdout.strip():
logger.debug(f"Claude CLI 本地调用成功: {len(result.stdout)} 字符")
return result.stdout.strip()
except FileNotFoundError:
logger.debug("本地 Claude CLI 未找到,尝试其他方式")
except subprocess.TimeoutExpired:
logger.warning("本地 Claude CLI 超时,尝试其他方式")
except Exception as e:
logger.debug(f"本地 Claude CLI 失败: {e}")
# ========== 方式2: SSH 调用宿主机 Claude CLI(Docker 容器内使用) ==========
try:
response = self._call_claude_via_ssh(prompt)
if response:
logger.debug(f"Claude CLI SSH 调用成功: {len(response)} 字符")
return response
except Exception as e:
logger.debug(f"SSH 调用 Claude CLI 失败: {e}")
if result.returncode == 0 and stdout:
logger.debug(f"Claude CLI 响应成功: {len(stdout)} 字符")
return stdout
# ========== 方式3: 容器内 Claude CLI(如果已安装) ==========
try:
if not is_windows:
result = subprocess.run(
['claude', '--print'],
input=prompt,
capture_output=True,
text=True,
timeout=self.timeout,
env=env,
encoding='utf-8',
errors='replace'
)
if result.returncode == 0 and result.stdout and result.stdout.strip():
logger.debug(f"Claude CLI 容器内调用成功: {len(result.stdout)} 字符")
return result.stdout.strip()
except Exception as e:
logger.debug(f"容器内 Claude CLI 失败: {e}")
# 所有方式都失败
raise Exception("Claude CLI 不可用(本地/SSH/容器均无法调用)")
def _call_claude_via_ssh(self, prompt: str) -> str:
"""
通过 SSH 调用宿主机的 Claude CLI(适用于 Docker 容器内环境)
容器内通过宿主机网关 IP 访问宿主机 SSH 服务
Args:
prompt: Prompt 文本
Returns:
str: Claude 响应文本
Raises:
Exception: SSH 调用失败时抛出异常
"""
import tempfile
# 失败情况
error_msg = stderr or f"返回码 {result.returncode}"
if not stdout:
error_msg = "响应为空"
raise Exception(f"Claude CLI 错误: {error_msg}")
# 获取宿主机网关 IP(Docker 容器内访问宿主机的方式)
host_ip = os.environ.get('CLAUDE_SSH_HOST', '172.17.0.1') # Docker 默认网关
host_port = int(os.environ.get('CLAUDE_SSH_PORT', '22'))
host_user = os.environ.get('CLAUDE_SSH_USER', 'ubains')
host_pass = os.environ.get('CLAUDE_SSH_PASS', 'Ubains@123')
logger.info(f"通过 SSH 调用宿主机 Claude CLI: {host_user}@{host_ip}:{host_port}")
# 将 prompt 写入临时文件,避免 shell 转义问题
with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False, encoding='utf-8') as f:
f.write(prompt)
prompt_file = f.name
try:
# 使用 SSH 执行 claude --print,通过 stdin 传递 prompt
ssh_cmd = [
'sshpass', '-p', host_pass,
'ssh', '-o', 'StrictHostKeyChecking=no',
'-o', 'ConnectTimeout=5',
'-p', str(host_port),
f'{host_user}@{host_ip}',
'claude --print'
]
result = subprocess.run(
ssh_cmd,
input=prompt,
capture_output=True,
text=True,
timeout=self.timeout,
encoding='utf-8',
errors='replace'
)
if result.returncode == 0 and result.stdout and result.stdout.strip():
return result.stdout.strip()
# sshpass 不可用,尝试 paramiko
if 'sshpass' in (result.stderr or ''):
logger.debug("sshpass 不可用,尝试 paramiko")
return self._call_claude_via_paramiko(prompt, host_ip, host_port, host_user, host_pass)
error_msg = result.stderr.strip() if result.stderr else "未知错误"
raise Exception(f"SSH Claude CLI 错误: {error_msg[:200]}")
except subprocess.TimeoutExpired:
raise Exception(f"Claude CLI 超时(>{self.timeout}s)")
raise Exception(f"SSH Claude CLI 超时(>{self.timeout}s)")
except FileNotFoundError:
raise Exception("Claude CLI 未安装或不在 PATH 中")
# sshpass 不存在,回退到 paramiko
logger.debug("sshpass 未安装,尝试 paramiko")
return self._call_claude_via_paramiko(prompt, host_ip, host_port, host_user, host_pass)
finally:
# 清理临时文件
try:
os.unlink(prompt_file)
except Exception:
pass
def _call_claude_via_paramiko(self, prompt: str, host: str, port: int, user: str, password: str) -> str:
"""
通过 paramiko SSH 库调用宿主机 Claude CLI
Args:
prompt: Prompt 文本
host: 宿主机 IP
port: SSH 端口
user: SSH 用户名
password: SSH 密码
Returns:
str: Claude 响应文本
"""
try:
import paramiko
except ImportError:
raise Exception("paramiko 未安装,无法通过 SSH 调用 Claude CLI")
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
ssh.connect(host, port=port, username=user, password=password, timeout=10)
# 通过 stdin 传递 prompt
stdin, stdout, stderr = ssh.exec_command(
'claude --print',
timeout=self.timeout
)
stdin.write(prompt)
stdin.flush()
stdin.channel.shutdown_write()
out = stdout.read().decode('utf-8', errors='replace').strip()
err = stderr.read().decode('utf-8', errors='replace').strip()
if out:
return out
raise Exception(f"Paramiko Claude CLI 无输出: {err[:200]}")
except Exception as e:
raise Exception(f"Claude CLI 调用失败: {e}")
raise Exception(f"Paramiko SSH 调用失败: {e}")
finally:
ssh.close()
def _parse_response(self, response: str) -> dict:
"""
......
Markdown 格式
0%
您添加了 0 到此讨论。请谨慎行事。
请先完成此评论的编辑!
注册 或者 后发表评论