"""FastAPI 主入口。""" import asyncio import logging import os from contextlib import asynccontextmanager from typing import Any import inngest import inngest.fast_api import uvicorn from fastapi import FastAPI from pydantic import BaseModel from butler.config import settings from butler.gateway.base import BaseGateway from butler.gateway.napcat import NapCatGateway from butler.gateway.telegram import TelegramGateway from butler.session.context import ContextManager from butler.workflows import inngest_client from butler.workflows.handle_message import handle_im_message from butler.workflows.guardrail import command_guardrail logging.basicConfig( level=getattr(logging, settings.log_level), format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", ) logger = logging.getLogger(__name__) class GatewayManager: """多通道网关管理器。""" def __init__(self) -> None: self.gateways: dict[str, BaseGateway] = {} self._tasks: list[asyncio.Task[None]] = [] def register(self, gateway: BaseGateway) -> None: """注册网关。""" self.gateways[gateway.name] = gateway async def start_all(self) -> None: """启动所有网关。""" for name, gw in self.gateways.items(): gw.on_message(self._handle_message) task = asyncio.create_task(gw.start()) self._tasks.append(task) logger.info(f"Gateway '{name}' started") async def _handle_message(self, user_id: str, content: str, channel: str) -> None: """统一消息回调：处理并回复。""" logger.info(f"[{channel}] Message from {user_id}: {content[:50]}...") try: result = await _process_message(user_id, content) reply = result.get("reply", "") if reply and channel in self.gateways: await self.gateways[channel].send_message(user_id, reply) logger.info(f"[{channel}] Replied to {user_id}") except Exception as e: logger.error(f"[{channel}] Error handling message: {e}") if channel in self.gateways: await self.gateways[channel].send_message(user_id, f"处理出错：{e}") async def stop_all(self) -> None: """停止所有网关。""" for name, gw in self.gateways.items(): await gw.stop() logger.info(f"Gateway '{name}' stopped") for task in self._tasks: task.cancel() def _build_gateways() -> GatewayManager: """根据配置构建网关实例。""" mgr = GatewayManager() for channel in settings.enabled_channels: if channel == "telegram" and settings.telegram_bot_token: mgr.register(TelegramGateway( bot_token=settings.telegram_bot_token, allowed_chats=settings.telegram_allowed_chats, )) elif channel == "napcat": mgr.register(NapCatGateway( ws_url=settings.napcat_ws_url, access_token=settings.napcat_token, )) else: logger.warning(f"Unknown channel: {channel}") return mgr gateway_manager = _build_gateways() context_manager = ContextManager() HELP_TEXT = """\ Butler-Shell 使用说明： !<命令> 原始命令模式，跳过 LLM，直接执行并返回原始输出 例：!ls -la, !df -h /new 新建会话（新 tmux session + 新 LLM 上下文） /list 列出所有会话 /enter <会话> 切换到指定会话（按编号或名称） /rename <名称> 重命名当前会话 /del <会话> 删除指定会话（按编号或名称） /clear 清理当前会话的 LLM 对话历史（保留 shell session） /help 显示本帮助信息 其他 自然语言模式，LLM 理解意图并总结输出 例：查看当前目录, 看下磁盘空间\ """ async def _process_message(user_id: str, content: str) -> dict[str, Any]: """核心消息处理逻辑：命令路由 → 执行 → 总结。""" from butler.session.wrapper import TmuxWrapper from butler.session import SessionMode from butler.security.guardrail import is_dangerous_command, get_danger_reason from butler.agent import agent from butler.skills.log_skill import LogSkill tmux = TmuxWrapper(session_prefix=settings.session_prefix) log_skill = LogSkill() # 检查权限 if not settings.is_user_allowed(user_id): return {"reply": f"用户 {user_id} 无权限使用此服务", "mode": "unauthorized", "command": "", "raw_output": "", "session_id": ""} # ---- 命令路由 ---- # /help if content.strip() == "/help": return {"reply": HELP_TEXT, "mode": "command", "command": "/help", "raw_output": "", "session_id": ""} # /new: 新建会话 if content.strip() == "/new": ctx = context_manager.get(user_id) new_seq = (ctx.seq + 1) if ctx else 1 old_session = ctx.session_id if ctx else None new_session_id = tmux.get_or_create_session(user_id, seq=new_seq) context_manager.new_session(user_id, new_session_id, new_seq) # 旧会话保留，不主动销毁 reply = f"已创建新会话: {new_session_id} (#{new_seq})" if old_session: reply += f"\n旧会话 {old_session} 仍保留" return {"reply": reply, "mode": "command", "command": "/new", "raw_output": "", "session_id": new_session_id} # /clear: 清理 LLM 上下文 if content.strip() == "/clear": ctx = context_manager.get(user_id) if ctx is None: # 还没有上下文，先创建 session_name = tmux.get_or_create_session(user_id) ctx = context_manager.get_or_create(user_id, session_name) context_manager.clear_context(user_id) return {"reply": f"已清理 LLM 对话历史 (会话: {ctx.session_id})", "mode": "command", "command": "/clear", "raw_output": "", "session_id": ctx.session_id} # /list: 列出所有会话 if content.strip() == "/list": sessions = context_manager.list_sessions(user_id) active_id = context_manager.get_active_session_id(user_id) if not sessions: return {"reply": "暂无会话，发送消息或 /new 创建", "mode": "command", "command": "/list", "raw_output": "", "session_id": ""} lines = ["会话列表："] for s in sessions: marker = " *" if s.session_id == active_id else "" name_part = f" ({s.name})" if s.name else "" lines.append(f" #{s.seq} {s.session_id}{name_part}{marker}") return {"reply": "\n".join(lines), "mode": "command", "command": "/list", "raw_output": "", "session_id": active_id} # /enter <会话>: 切换到指定会话 if content.strip().startswith("/enter "): target = content.strip()[7:].strip() sessions = context_manager.list_sessions(user_id) if not sessions: return {"reply": "暂无会话可切换", "mode": "command", "command": "/enter", "raw_output": "", "session_id": ""} # 按 seq 匹配 matched = None for s in sessions: if str(s.seq) == target or s.name == target or s.session_id == target: matched = s break if matched is None: return {"reply": f"未找到会话: {target}。发送 /list 查看所有会话", "mode": "command", "command": "/enter", "raw_output": "", "session_id": ""} if not tmux.session_exists(matched.session_id): return {"reply": f"会话 {matched.session_id} 的 tmux session 已不存在", "mode": "command", "command": "/enter", "raw_output": "", "session_id": ""} context_manager.switch_session(user_id, matched.session_id) return {"reply": f"已切换到会话: {matched.display_name}", "mode": "command", "command": "/enter", "raw_output": "", "session_id": matched.session_id} # /rename <名称>: 重命名当前会话 if content.strip().startswith("/rename "): new_name = content.strip()[8:].strip() if not new_name: return {"reply": "用法: /rename <名称>", "mode": "command", "command": "/rename", "raw_output": "", "session_id": ""} ctx = context_manager.get(user_id) if ctx is None: session_name = tmux.get_or_create_session(user_id) ctx = context_manager.get_or_create(user_id, session_name) context_manager.rename_session(user_id, ctx.session_id, new_name) return {"reply": f"已重命名会话为: {new_name}", "mode": "command", "command": "/rename", "raw_output": "", "session_id": ctx.session_id} # /del <会话>: 删除指定会话 if content.strip().startswith("/del "): target = content.strip()[5:].strip() sessions = context_manager.list_sessions(user_id) matched = None for s in sessions: if str(s.seq) == target or s.name == target or s.session_id == target: matched = s break if matched is None: return {"reply": f"未找到会话: {target}", "mode": "command", "command": "/del", "raw_output": "", "session_id": ""} active_id = context_manager.get_active_session_id(user_id) if matched.session_id == active_id and len(sessions) <= 1: return {"reply": "不能删除唯一活跃会话，请先 /new 创建新会话", "mode": "command", "command": "/del", "raw_output": "", "session_id": ""} context_manager.delete_session(user_id, matched.session_id) tmux.kill_session(matched.session_id) new_active = context_manager.get_active_session_id(user_id) return {"reply": f"已删除会话: {matched.display_name}，当前会话: {new_active}", "mode": "command", "command": "/del", "raw_output": "", "session_id": new_active} # ! 前缀：原始命令模式 if content.startswith("!"): command = content[1:].strip() if not command: return {"reply": "命令不能为空", "mode": "error", "command": "", "raw_output": "", "session_id": ""} # 获取或创建会话 ctx = context_manager.get_or_create(user_id, "") if not ctx.session_id or not tmux.session_exists(ctx.session_id): session_name = tmux.get_or_create_session(user_id, seq=ctx.seq) ctx.session_id = session_name # 危险命令检查 if settings.guardrail_enabled and is_dangerous_command(command): reason = get_danger_reason(command) or "潜在危险命令" return { "reply": f"⚠️ 该操作需要确认：{reason}。将执行 `{command}`，确认吗？", "mode": "pending_approval", "command": command, "raw_output": "", "session_id": ctx.session_id, } # 执行命令 tmux.send_keys(ctx.session_id, command, enter=True) await asyncio.sleep(0.5) raw_output = tmux.capture_pane(ctx.session_id) # 记录日志 log_skill.log_command(user_id, ctx.session_id, command, raw_output[:2000]) return { "reply": raw_output[-2000:], "mode": "raw", "command": command, "raw_output": raw_output[-2000:], "session_id": ctx.session_id, } # ---- 自然语言模式 ---- # 获取或创建会话上下文 session_name = tmux.get_or_create_session(user_id) ctx = context_manager.get_or_create(user_id, session_name) if ctx.session_id != session_name: ctx.session_id = session_name # 检测模式 current_mode = tmux.detect_mode(session_name) # 交互模式直接透传 if current_mode == SessionMode.INTERACTIVE: tmux.send_keys(session_name, content, enter=False) raw_output = tmux.capture_pane(session_name) return {"reply": "已发送按键到交互程序", "mode": "interactive", "command": content, "raw_output": raw_output[-2000:], "session_id": session_name} # 自然语言模式：LLM 理解意图 → 生成命令 command = content llm_failed = False if agent.is_available(): context_info = { "pwd": tmux.capture_pane(session_name)[-500:] if tmux.session_exists(session_name) else "", } command = await agent.understand(content, context_info, llm_messages=ctx.llm_messages) logger.info(f"Agent: '{content}' -> '{command}'") # LLM 返回了原始输入 = 调用失败，不执行 if command == content: llm_failed = True if llm_failed: return { "reply": f"⚠️ LLM 服务暂时不可用，未能理解指令。请稍后重试。\n原始输入: {content}", "mode": "llm_error", "command": "", "raw_output": "", "session_id": session_name, } # 危险命令检查 if settings.guardrail_enabled and is_dangerous_command(command): reason = get_danger_reason(command) or "潜在危险命令" return { "reply": f"⚠️ 该操作需要确认：{reason}。将执行 `{command}`，确认吗？", "mode": "pending_approval", "command": command, "raw_output": "", "session_id": session_name, } # 执行命令 tmux.send_keys(session_name, command, enter=True) await asyncio.sleep(0.5) raw_output = tmux.capture_pane(session_name) # 记录日志 log_skill.log_command(user_id, session_name, command, raw_output[:2000]) # 记录到 LLM 上下文 context_manager.add_message(user_id, "user", content) context_manager.add_message(user_id, "assistant", f"执行了: {command}\n输出摘要待生成") # LLM 总结输出 reply = "" if agent.is_available(): reply = await agent.summarize(content, command, raw_output, llm_messages=ctx.llm_messages) if not reply: reply = f"⚠️ 总结服务暂时不可用，以下是命令原始输出:\n执行了 `{command}`\n{raw_output[-500:]}" # 更新 LLM 上下文中的 assistant 消息 if ctx.llm_messages: ctx.llm_messages[-1]["content"] = reply else: reply = f"执行了 `{command}`\n{raw_output[-500:]}" return {"reply": reply, "mode": "nl", "command": command, "raw_output": raw_output[-2000:], "session_id": session_name} @asynccontextmanager async def lifespan(app: FastAPI): """应用生命周期管理。""" logger.info("Starting Butler-Shell...") await gateway_manager.start_all() yield logger.info("Shutting down Butler-Shell...") await gateway_manager.stop_all() app = FastAPI( title="Butler-Shell", description="Persistent Shell Assistant with Multi-Channel Bridge", version="0.2.0", lifespan=lifespan, ) # 注册 Inngest 函数 _signing_key = settings.inngest_signing_key or os.environ.get( "INNGEST_SIGNING_KEY", "signkey-dev-placeholder-1234567890123456789012345678901234567890" ) _dev_client = inngest.Inngest( app_id="butler-shell", event_key=settings.inngest_event_key or "dev_key", signing_key=_signing_key, is_production=bool(settings.inngest_signing_key), ) inngest.fast_api.serve( app, _dev_client, [handle_im_message, command_guardrail], serve_path="/api/inngest", ) @app.get("/health") async def health() -> dict[str, str]: """服务存活检查。""" return {"status": "ok"} @app.get("/ready") async def ready() -> dict[str, Any]: """服务就绪检查。""" import subprocess try: subprocess.run(["tmux", "-V"], capture_output=True, check=True) tmux_ok = True except Exception: tmux_ok = False channels = {name: "connected" for name in gateway_manager.gateways} return { "status": "ready" if tmux_ok else "degraded", "tmux": tmux_ok, "channels": channels, } class MockMessage(BaseModel): """Mock 消息请求。""" user_id: str = "test_user_001" content: str @app.post("/api/mock/message") async def mock_message(msg: MockMessage) -> dict[str, Any]: """自然语言交互端点：输入自然语言，返回自然语言总结。""" return await _process_message(msg.user_id, msg.content) @app.get("/api/raw-output") async def get_raw_output(user_id: str, limit: int = 5) -> dict[str, Any]: """获取命令的原始输出（从日志）。""" from butler.skills.log_skill import LogSkill log_skill = LogSkill() conn = __import__("sqlite3").connect(log_skill.db_path) cursor = conn.execute( "SELECT command, output, timestamp FROM shell_logs WHERE user_id = ? ORDER BY timestamp DESC LIMIT ?", (user_id, limit), ) rows = cursor.fetchall() conn.close() logs = [ {"command": row[0], "output": row[1], "timestamp": row[2]} for row in rows ] return {"user_id": user_id, "logs": logs} @app.post("/api/approval") async def handle_approval(user_id: str, approved: bool) -> dict[str, str]: """处理用户审批响应。""" await inngest_client.send( inngest.Event( name="im/approval-response", data={"user_id": user_id, "approved": approved}, ) ) return {"status": "sent"} def main() -> None: """启动服务。""" uvicorn.run("butler.main:app", host="0.0.0.0", port=9600, reload=False) if __name__ == "__main__": main()