568dcc45e4
- Remove duplicate claude_api provider to fix automatic failover - Enhance error detection with HTTP status codes and more indicators - Add comprehensive README documentation with manual switching - Implement Discord bot with Claude Code CLI integration - Support /terminal and /claude commands with AI-powered responses 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
111 lines
4.1 KiB
Python
111 lines
4.1 KiB
Python
import asyncio
|
||
import logging
|
||
import aiohttp
|
||
import json
|
||
from typing import Optional, Dict, List
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
class ClaudeClient:
|
||
"""Claude API客户端"""
|
||
|
||
def __init__(self, api_key: str, base_url: str = "https://api.anthropic.com"):
|
||
self.api_key = api_key
|
||
self.base_url = base_url
|
||
self.model = "claude-3-5-sonnet-20241022"
|
||
|
||
async def chat_completion(self, messages: List[Dict], system_prompt: str = None, max_tokens: int = 1000) -> Optional[str]:
|
||
"""发送聊天完成请求到Claude API"""
|
||
try:
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"x-api-key": self.api_key,
|
||
"anthropic-version": "2023-06-01"
|
||
}
|
||
|
||
data = {
|
||
"model": self.model,
|
||
"max_tokens": max_tokens,
|
||
"messages": messages
|
||
}
|
||
|
||
if system_prompt:
|
||
data["system"] = system_prompt
|
||
|
||
async with aiohttp.ClientSession() as session:
|
||
async with session.post(
|
||
f"{self.base_url}/v1/messages",
|
||
headers=headers,
|
||
json=data,
|
||
timeout=aiohttp.ClientTimeout(total=30)
|
||
) as response:
|
||
|
||
if response.status == 200:
|
||
result = await response.json()
|
||
return result["content"][0]["text"]
|
||
else:
|
||
error_text = await response.text()
|
||
logger.error(f"Claude API错误 {response.status}: {error_text}")
|
||
return None
|
||
|
||
except asyncio.TimeoutError:
|
||
logger.error("Claude API请求超时")
|
||
return None
|
||
except Exception as e:
|
||
logger.error(f"Claude API请求失败: {e}")
|
||
return None
|
||
|
||
async def interpret_command(self, command: str, context: Dict = None) -> Optional[str]:
|
||
"""让Claude解释并转换自然语言命令为shell命令"""
|
||
|
||
system_prompt = """你是一个Docker服务器管理助手。用户会给你自然语言命令，你需要将其转换为安全的shell命令。
|
||
|
||
工作目录: /home/will/docker
|
||
允许的操作: ls, pwd, cat, grep, find, docker, docker-compose, tail, head, ps, df, du
|
||
禁止的操作: rm -rf, chmod 777, 任何可能破坏系统的命令
|
||
|
||
如果是危险命令或超出范围的操作，回复"DENIED: 原因"
|
||
如果是正常命令，只回复shell命令，不要解释。
|
||
|
||
示例:
|
||
用户: "查看Docker容器状态"
|
||
你: "docker ps"
|
||
|
||
用户: "删除所有文件"
|
||
你: "DENIED: 危险操作，可能删除重要文件"
|
||
"""
|
||
|
||
messages = [
|
||
{"role": "user", "content": f"请将这个命令转换为shell命令: {command}"}
|
||
]
|
||
|
||
if context:
|
||
# 添加上下文信息
|
||
context_info = f"当前目录上下文: {context.get('current_dir', '/home/will/docker')}"
|
||
if context.get('last_commands'):
|
||
context_info += f"\n最近执行的命令: {', '.join(context['last_commands'][-3:])}"
|
||
|
||
messages[0]["content"] = f"{context_info}\n\n{messages[0]['content']}"
|
||
|
||
return await self.chat_completion(messages, system_prompt, max_tokens=150)
|
||
|
||
async def explain_error(self, command: str, error_output: str) -> Optional[str]:
|
||
"""让Claude解释错误并提供解决建议"""
|
||
|
||
system_prompt = """你是一个Docker服务器管理助手。用户执行命令时遇到错误，请简洁地解释问题并提供解决建议。
|
||
|
||
要求:
|
||
1. 简洁明了，不超过3行
|
||
2. 重点说明可能的原因
|
||
3. 提供具体的解决方案
|
||
4. 如果是权限问题，提醒用户当前限制
|
||
"""
|
||
|
||
messages = [
|
||
{
|
||
"role": "user",
|
||
"content": f"命令: {command}\n错误输出: {error_output}\n\n请解释这个错误并提供解决建议。"
|
||
}
|
||
]
|
||
|
||
return await self.chat_completion(messages, system_prompt, max_tokens=200) |