import { Bot, Context } from "grammy"; import Anthropic from "@anthropic-ai/sdk"; import { readFileSync } from "fs"; import { resolve } from "path"; // ─── Config ─── const BOT_TOKEN = process.env.TELEGRAM_BOT_TOKEN!; const ANTHROPIC_KEY = process.env.ANTHROPIC_API_KEY!; const SSH_USER = process.env.SSH_USER || "root"; const SSH_HOST = process.env.SSH_HOST!; const SSH_PORT = process.env.SSH_PORT || "22"; const ALLOWED_USERS = process.env.TELEGRAM_ALLOWED_USERS?.split(",").map(Number) || []; if (!BOT_TOKEN) throw new Error("TELEGRAM_BOT_TOKEN missing from .env"); if (!ANTHROPIC_KEY) throw new Error("ANTHROPIC_API_KEY missing from .env"); if (!SSH_HOST) throw new Error("SSH_HOST missing from .env"); // ─── Load infra context ─── let infraContext = ""; try { infraContext = readFileSync(resolve(__dirname, "../.pi/infra.md"), "utf-8"); } catch { console.warn("⚠️ Could not load .pi/infra.md"); } // ─── Init ─── const bot = new Bot(BOT_TOKEN); const anthropic = new Anthropic({ apiKey: ANTHROPIC_KEY }); // Per-user conversation history const conversations = new Map(); const MAX_HISTORY = 30; // ─── Auth middleware ─── function isAuthorized(ctx: Context): boolean { if (ALLOWED_USERS.length === 0) return true; // no whitelist = open return ALLOWED_USERS.includes(ctx.from?.id || 0); } // ─── SSH exec helper ─── async function sshExec(command: string, timeoutSec = 30): Promise { const proc = Bun.spawn( ["ssh", "-o", "ConnectTimeout=5", "-o", "StrictHostKeyChecking=accept-new", "-p", SSH_PORT, `${SSH_USER}@${SSH_HOST}`, command], { stdout: "pipe", stderr: "pipe" } ); const timer = setTimeout(() => proc.kill(), timeoutSec * 1000); const [stdout, stderr] = await Promise.all([ new Response(proc.stdout).text(), new Response(proc.stderr).text(), ]); clearTimeout(timer); await proc.exited; const output = (stdout + stderr).trim(); return output.length > 3500 ? output.slice(0, 3500) + "\n...(truncated)" : output; } // ─── System prompt ─── const SYSTEM_PROMPT = `You are a DevOps assistant connected to a live server via Telegram. You have access to run SSH commands on the server. You prioritize uptime. Infrastructure context: ${infraContext} RULES: - If the user asks to run a command, use the run_command tool. - For destructive commands (rm -rf, DROP, reboot, etc.), WARN the user and ask for confirmation before executing. - Keep responses concise — this is Telegram, not a terminal. - You can chain multiple commands in one SSH call using && or ; - Format output with monospace blocks for readability.`; // ─── Claude tools ─── const tools: Anthropic.Tool[] = [ { name: "run_command", description: "Execute a command on the server via SSH. Use for any server inspection, management, or debugging task.", input_schema: { type: "object" as const, properties: { command: { type: "string", description: "Shell command to run on the server" }, timeout: { type: "number", description: "Timeout in seconds (default 30)" }, }, required: ["command"], }, }, ]; // ─── Process message through Claude ─── async function chat(userId: number, message: string): Promise { // Get or init conversation history if (!conversations.has(userId)) conversations.set(userId, []); const history = conversations.get(userId)!; history.push({ role: "user", content: message }); // Trim history if too long while (history.length > MAX_HISTORY) history.shift(); let response = await anthropic.messages.create({ model: "claude-sonnet-4-20250514", max_tokens: 1024, system: SYSTEM_PROMPT, tools, messages: history, }); // Tool use loop while (response.stop_reason === "tool_use") { const assistantContent = response.content; history.push({ role: "assistant", content: assistantContent }); const toolResults: Anthropic.ToolResultBlockParam[] = []; for (const block of assistantContent) { if (block.type === "tool_use") { const input = block.input as { command: string; timeout?: number }; console.log(`🔧 [${userId}] run_command: ${input.command}`); try { const output = await sshExec(input.command, input.timeout || 30); toolResults.push({ type: "tool_result", tool_use_id: block.id, content: output || "(no output)" }); } catch (err: any) { toolResults.push({ type: "tool_result", tool_use_id: block.id, content: `ERROR: ${err.message}`, is_error: true }); } } } history.push({ role: "user", content: toolResults }); response = await anthropic.messages.create({ model: "claude-sonnet-4-20250514", max_tokens: 1024, system: SYSTEM_PROMPT, tools, messages: history, }); } // Extract text response const text = response.content .filter((b): b is Anthropic.TextBlock => b.type === "text") .map((b) => b.text) .join("\n"); history.push({ role: "assistant", content: text }); return text || "(no response)"; } // ─── Bot handlers ─── bot.command("start", (ctx) => { if (!isAuthorized(ctx)) return ctx.reply("⛔ Not authorized."); return ctx.reply( "🤖 *CR Management Bot*\n\n" + "I'm connected to your server. You can:\n" + "• Ask me anything about your infra\n" + "• Ask me to run commands\n" + "• `/status` — quick server health\n" + "• `/clear` — reset conversation\n" + "• `/ssh ` — run a command directly\n", { parse_mode: "Markdown" } ); }); bot.command("status", async (ctx) => { if (!isAuthorized(ctx)) return ctx.reply("⛔ Not authorized."); await ctx.reply("🔍 Checking server..."); try { const output = await sshExec("uptime && echo '---' && df -h / && echo '---' && free -h | head -2 && echo '---' && incus list --format csv -c ns 2>/dev/null"); return ctx.reply(`\`\`\`\n${output}\n\`\`\``, { parse_mode: "Markdown" }); } catch (err: any) { return ctx.reply(`❌ SSH failed: ${err.message}`); } }); bot.command("ssh", async (ctx) => { if (!isAuthorized(ctx)) return ctx.reply("⛔ Not authorized."); const cmd = ctx.match; if (!cmd) return ctx.reply("Usage: `/ssh `", { parse_mode: "Markdown" }); await ctx.reply(`⚡ Running: \`${cmd}\``, { parse_mode: "Markdown" }); try { const output = await sshExec(cmd); return ctx.reply(`\`\`\`\n${output || "(no output)"}\n\`\`\``, { parse_mode: "Markdown" }); } catch (err: any) { return ctx.reply(`❌ ${err.message}`); } }); bot.command("clear", (ctx) => { if (!isAuthorized(ctx)) return ctx.reply("⛔ Not authorized."); conversations.delete(ctx.from!.id); return ctx.reply("🧹 Conversation cleared."); }); // Free text → Claude bot.on("message:text", async (ctx) => { if (!isAuthorized(ctx)) return ctx.reply("⛔ Not authorized."); const userId = ctx.from!.id; const msg = ctx.message!.text; await ctx.replyWithChatAction("typing"); try { const reply = await chat(userId, msg); // Split long messages (Telegram 4096 char limit) const chunks = reply.match(/[\s\S]{1,4000}/g) || ["(empty)"]; for (const chunk of chunks) { await ctx.reply(chunk, { parse_mode: "Markdown" }).catch(() => ctx.reply(chunk) // retry without markdown if it fails ); } } catch (err: any) { console.error(`❌ [${userId}]`, err); await ctx.reply(`❌ Error: ${err.message}`); } }); // ─── Launch ─── console.log("🚀 Bot starting..."); bot.start({ onStart: (info) => console.log(`✅ @${info.username} is live`), });