Plantillas de inicio de bots en Node.js y Python (Telegram, Discord, Slack, WhatsApp)

import { Bot } from "grammy";

const bot = new Bot(process.env.BOT_TOKEN);

bot.command("start", (ctx) =>
  ctx.reply("Hello! Send me any text and I will echo it. /help lists commands."));
bot.command("help", (ctx) => ctx.reply("/start - greeting\n/help - this message"));

bot.on("message:text", (ctx) => ctx.reply(ctx.message.text));

bot.catch((err) => console.error("Handler error:", err.error));

// Long polling: no public URL needed. Stops cleanly on Ctrl+C.
bot.start({ onStart: (me) => console.log(`Running as @${me.username}`) });
process.once("SIGINT", () => bot.stop());
process.once("SIGTERM", () => bot.stop());

import logging
import os

from telegram import Update
from telegram.ext import Application, CommandHandler, ContextTypes, MessageHandler, filters

logging.basicConfig(level=logging.INFO)


async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    await update.message.reply_text("Hello! Send me any text and I will echo it.")


async def echo(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    await update.message.reply_text(update.message.text)


def main() -> None:
    app = Application.builder().token(os.environ["BOT_TOKEN"]).build()
    app.add_handler(CommandHandler("start", start))
    app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, echo))
    app.run_polling(allowed_updates=Update.ALL_TYPES)


if __name__ == "__main__":
    main()

export default {
  async fetch(request, env, ctx) {
    if (request.method !== "POST") return new Response("ok");
    // Telegram sends the secret you passed to setWebhook in this header.
    if (request.headers.get("X-Telegram-Bot-Api-Secret-Token") !== env.WEBHOOK_SECRET) {
      return new Response("forbidden", { status: 403 });
    }
    const update = await request.json();
    const msg = update.message;
    if (msg?.text) {
      // Reply after returning 200 so Telegram does not retry.
      ctx.waitUntil(send(env, msg.chat.id, msg.text === "/start" ? "Hello from a Worker!" : msg.text));
    }
    return new Response("ok");
  },
};

async function send(env, chatId, text) {
  const res = await fetch(`https://api.telegram.org/bot${env.BOT_TOKEN}/sendMessage`, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ chat_id: chatId, text }),
  });
  if (!res.ok) console.error("sendMessage failed", res.status, await res.text());
}

import { Client, Events, GatewayIntentBits, MessageFlags, REST, Routes, SlashCommandBuilder } from "discord.js";

const { DISCORD_TOKEN, CLIENT_ID, GUILD_ID } = process.env;

const commands = [
  new SlashCommandBuilder().setName("ping").setDescription("Check that the bot is alive").toJSON(),
];

// Guild commands update instantly; use Routes.applicationCommands for global ones.
await new REST().setToken(DISCORD_TOKEN).put(Routes.applicationGuildCommands(CLIENT_ID, GUILD_ID), { body: commands });

const client = new Client({ intents: [GatewayIntentBits.Guilds] });

client.once(Events.ClientReady, (c) => console.log(`Ready as ${c.user.tag}`));

client.on(Events.InteractionCreate, async (interaction) => {
  if (!interaction.isChatInputCommand()) return;
  if (interaction.commandName === "ping") {
    await interaction.reply({ content: `Pong (${client.ws.ping} ms)`, flags: MessageFlags.Ephemeral });
  }
});

client.login(DISCORD_TOKEN);

import os

import discord
from discord import app_commands

intents = discord.Intents.default()
client = discord.Client(intents=intents)
tree = app_commands.CommandTree(client)
GUILD = discord.Object(id=int(os.environ["GUILD_ID"]))


@tree.command(name="ping", description="Check that the bot is alive", guild=GUILD)
async def ping(interaction: discord.Interaction) -> None:
    await interaction.response.send_message(f"Pong ({round(client.latency * 1000)} ms)", ephemeral=True)


@client.event
async def on_ready() -> None:
    await tree.sync(guild=GUILD)
    print(f"Ready as {client.user}")


client.run(os.environ["DISCORD_TOKEN"])

import bolt from "@slack/bolt";

const { App } = bolt;

const app = new App({
  token: process.env.SLACK_BOT_TOKEN,
  appToken: process.env.SLACK_APP_TOKEN,
  socketMode: true,
});

app.message("hello", async ({ message, say }) => {
  await say(`Hi <@${message.user}>!`);
});

app.command("/status", async ({ ack, respond }) => {
  await ack(); // acknowledge within 3 seconds
  await respond("All systems normal.");
});

await app.start();
console.log("Bolt app running in Socket Mode");

import os

from slack_bolt import App
from slack_bolt.adapter.socket_mode import SocketModeHandler

app = App(token=os.environ["SLACK_BOT_TOKEN"])


@app.message("hello")
def hello(message, say):
    say(f"Hi <@{message['user']}>!")


@app.command("/status")
def status(ack, respond):
    ack()  # acknowledge within 3 seconds
    respond("All systems normal.")


if __name__ == "__main__":
    SocketModeHandler(app, os.environ["SLACK_APP_TOKEN"]).start()

import crypto from "node:crypto";
import http from "node:http";

const { VERIFY_TOKEN, APP_SECRET, WHATSAPP_TOKEN, PHONE_NUMBER_ID, PORT = 3000 } = process.env;
const GRAPH = "https://graph.facebook.com/v21.0"; // use the current Graph API version

http.createServer(async (req, res) => {
  const url = new URL(req.url, "http://localhost");
  if (url.pathname !== "/webhook") return res.writeHead(404).end();

  if (req.method === "GET") {
    const ok = url.searchParams.get("hub.mode") === "subscribe" && url.searchParams.get("hub.verify_token") === VERIFY_TOKEN;
    return res.writeHead(ok ? 200 : 403).end(ok ? url.searchParams.get("hub.challenge") : "");
  }

  const chunks = [];
  for await (const c of req) chunks.push(c);
  const raw = Buffer.concat(chunks);
  const expected = "sha256=" + crypto.createHmac("sha256", APP_SECRET).update(raw).digest("hex");
  const got = String(req.headers["x-hub-signature-256"] || "");
  if (got.length !== expected.length || !crypto.timingSafeEqual(Buffer.from(got), Buffer.from(expected))) {
    return res.writeHead(401).end();
  }
  res.writeHead(200).end(); // acknowledge first

  const body = JSON.parse(raw.toString("utf8"));
  for (const entry of body.entry ?? []) {
    for (const change of entry.changes ?? []) {
      for (const msg of change.value?.messages ?? []) {
        if (msg.type !== "text") continue;
        await fetch(`${GRAPH}/${PHONE_NUMBER_ID}/messages`, {
          method: "POST",
          headers: { Authorization: `Bearer ${WHATSAPP_TOKEN}`, "Content-Type": "application/json" },
          body: JSON.stringify({ messaging_product: "whatsapp", to: msg.from, type: "text", text: { body: `You said: ${msg.text.body}` } }),
        });
      }
    }
  }
}).listen(PORT, () => console.log(`Listening on :${PORT}`));

import hashlib
import hmac
import os

import httpx
from fastapi import BackgroundTasks, FastAPI, Request, Response

VERIFY_TOKEN = os.environ["VERIFY_TOKEN"]
APP_SECRET = os.environ["APP_SECRET"].encode()
WHATSAPP_TOKEN = os.environ["WHATSAPP_TOKEN"]
PHONE_NUMBER_ID = os.environ["PHONE_NUMBER_ID"]
GRAPH = "https://graph.facebook.com/v21.0"  # use the current Graph API version

app = FastAPI()


@app.get("/webhook")
def verify(request: Request) -> Response:
    p = request.query_params
    if p.get("hub.mode") == "subscribe" and p.get("hub.verify_token") == VERIFY_TOKEN:
        return Response(p.get("hub.challenge", ""))
    return Response(status_code=403)


async def reply(to: str, text: str) -> None:
    async with httpx.AsyncClient(timeout=10) as client:
        await client.post(
            f"{GRAPH}/{PHONE_NUMBER_ID}/messages",
            headers={"Authorization": f"Bearer {WHATSAPP_TOKEN}"},
            json={"messaging_product": "whatsapp", "to": to, "type": "text", "text": {"body": f"You said: {text}"}},
        )


@app.post("/webhook")
async def receive(request: Request, tasks: BackgroundTasks) -> Response:
    raw = await request.body()
    expected = "sha256=" + hmac.new(APP_SECRET, raw, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, request.headers.get("x-hub-signature-256", "")):
        return Response(status_code=401)
    body = await request.json()
    for entry in body.get("entry", []):
        for change in entry.get("changes", []):
            for msg in change.get("value", {}).get("messages", []):
                if msg.get("type") == "text":
                    tasks.add_task(reply, msg["from"], msg["text"]["body"])
    return Response(status_code=200)

# Telegram
BOT_TOKEN=
WEBHOOK_SECRET=

# Discord
DISCORD_TOKEN=
CLIENT_ID=
GUILD_ID=

# Slack
SLACK_BOT_TOKEN=
SLACK_APP_TOKEN=

# WhatsApp Cloud API
VERIFY_TOKEN=
APP_SECRET=
WHATSAPP_TOKEN=
PHONE_NUMBER_ID=