AI in messaging apps: Telegram, Slack, WhatsApp and Teams on autopilot

import os
import logging
from telegram import Update
from telegram.ext import (
    Application, CommandHandler, MessageHandler,
    filters, ContextTypes
)
import anthropic
from dotenv import load_dotenv

load_dotenv()

claude = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
BOT_TOKEN = os.environ["TELEGRAM_BOT_TOKEN"]

# Conversation history kept in memory
# In production, replace with Redis or SQLite
conversation_history: dict[int, list] = {}

SYSTEM_PROMPT = """You are the AI assistant for Acme Realty, a real estate agency.
You answer questions about real estate in Ecuador: prices, neighborhoods, the buying process.
Tone: a friendly expert, not a salesperson.
If a question needs a consultation with an agent, say so clearly and offer to take their contact info.
Keep it short: 1–3 sentences for simple questions, up to 5 paragraphs for complex ones.
Language: English."""


async def cmd_start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """The /start command: greet a new user."""
    user = update.effective_user
    await update.message.reply_text(
        f"Hi, {user.first_name}! 👋\n\n"
        "I can help with questions about real estate in Cuenca, Ecuador. "
        "Ask about prices, neighborhoods or the buying process, and I'll answer right away.\n\n"
        "If you'd like a personal consultation, just say so."
    )


async def cmd_contact(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """The /contact command: collect contact details."""
    await update.message.reply_text(
        "Leave your email or phone number, "
        "and an agent will get back to you within 2 hours during business hours."
    )
    # In a real bot, a ConversationHandler goes here to receive and save the contact


async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """The main handler for incoming messages."""
    user_id = update.effective_user.id
    user_text = update.message.text

    # Start the history if this is the first message
    if user_id not in conversation_history:
        conversation_history[user_id] = []

    # Add the user's message
    conversation_history[user_id].append({
        "role": "user",
        "content": user_text
    })

    # Limit the history to the last 10 messages
    if len(conversation_history[user_id]) > 10:
        conversation_history[user_id] = conversation_history[user_id][-10:]

    # Show "typing..." for a good user experience
    await context.bot.send_chat_action(
        chat_id=update.effective_chat.id,
        action="typing"
    )

    # Request to Claude (current models: the "What's current" page)
    response = claude.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=1024,
        system=SYSTEM_PROMPT,
        messages=conversation_history[user_id]
    )

    assistant_reply = "".join(b.text for b in response.content if b.type == "text")

    # Save the reply to the history
    conversation_history[user_id].append({
        "role": "assistant",
        "content": assistant_reply
    })

    await update.message.reply_text(assistant_reply)


async def handle_error(update: object, context: ContextTypes.DEFAULT_TYPE):
    """Log errors."""
    logging.error("Exception while handling update:", exc_info=context.error)


def main():
    app = Application.builder().token(BOT_TOKEN).build()

    app.add_handler(CommandHandler("start", cmd_start))
    app.add_handler(CommandHandler("contact", cmd_contact))
    app.add_handler(MessageHandler(
        filters.TEXT & ~filters.COMMAND, handle_message
    ))
    app.add_error_handler(handle_error)

    print("Telegram bot running (polling mode)...")
    app.run_polling(allowed_updates=Update.ALL_TYPES)


if __name__ == "__main__":
    main()

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

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


@app.event("app_mention")
def handle_mention(event, say):
    """Replies when someone addresses the bot with an @mention."""
    user_text = event["text"]

    # Strip the bot's @mention from the text
    clean_text = user_text.split(">", 1)[-1].strip() if ">" in user_text else user_text

    response = claude.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=512,
        system="You are the development team's assistant. Keep answers short and specific.",
        messages=[{"role": "user", "content": clean_text}]
    )

    say(f"<@{event['user']}> {''.join(b.text for b in response.content if b.type == 'text')}")


@app.command("/standup")
def handle_standup(ack, say, command):
    """
    /standup: generates a standup form.
    The user fills it in, and the bot collects the answers and makes a summary.
    """
    ack()
    say({
        "blocks": [
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"*Standup for {command['user_name']}*\n"
                            "Answer three questions:"
                }
            },
            {
                "type": "input",
                "block_id": "done",
                "element": {
                    "type": "plain_text_input",
                    "action_id": "done_input",
                    "placeholder": {"type": "plain_text", "text": "What did you do yesterday?"}
                },
                "label": {"type": "plain_text", "text": "Done"}
            }
        ]
    })


@app.command("/summarize")
def handle_summarize(ack, say, command):
    """
    /summarize [text]: summarizes a long message.
    Handy for long PR descriptions or documents in channels.
    """
    ack()
    text_to_summarize = command.get("text", "")

    if not text_to_summarize:
        say("Add the text after the command: `/summarize [your text]`")
        return

    response = claude.messages.create(
        model="claude-haiku-4-5",
        max_tokens=300,
        messages=[{
            "role": "user",
            "content": f"Summarize this text in 3–5 bullet points:\n\n{text_to_summarize}"
        }]
    )

    say(f"📋 *Summary:*\n{''.join(b.text for b in response.content if b.type == 'text')}")


if __name__ == "__main__":
    handler = SocketModeHandler(app, os.environ["SLACK_APP_TOKEN"])
    print("Slack bot running...")
    handler.start()

from flask import Flask, request
from twilio.twiml.messaging_response import MessagingResponse
from twilio.rest import Client
import anthropic
import os

app = Flask(__name__)
twilio_client = Client(
    os.environ["TWILIO_ACCOUNT_SID"],
    os.environ["TWILIO_AUTH_TOKEN"]
)
claude = anthropic.Anthropic()

# Conversation history (in production: Redis with a 24h TTL)
wa_conversations: dict[str, list] = {}

WA_SYSTEM = """You are a company's AI assistant on WhatsApp.
Keep replies short (WhatsApp isn't email).
If a question needs a person, say: 'I'm passing you to a specialist, expect a reply within an hour.'
Be friendly but professional."""


@app.route("/webhook/whatsapp", methods=["POST"])
def whatsapp_webhook():
    """Handles incoming WhatsApp messages."""
    from_number = request.form.get("From", "")
    body = request.form.get("Body", "").strip()

    if not body:
        return str(MessagingResponse())

    # Conversation history
    if from_number not in wa_conversations:
        wa_conversations[from_number] = []

    wa_conversations[from_number].append(
        {"role": "user", "content": body}
    )

    # Limit the context to the last 8 messages
    history = wa_conversations[from_number][-8:]

    # Claude generates a reply
    response = claude.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=512,
        system=WA_SYSTEM,
        messages=history
    )

    reply = "".join(b.text for b in response.content if b.type == "text")

    wa_conversations[from_number].append(
        {"role": "assistant", "content": reply}
    )

    # If Claude hands off to a person, notify someone inside the company
    if "specialist" in reply.lower() or "passing you" in reply.lower():
        notify_manager(from_number, body, reply)

    # Reply to the user
    resp = MessagingResponse()
    resp.message(reply)
    return str(resp)


def notify_manager(customer_number: str, question: str, bot_reply: str):
    """Notify a manager on Telegram when the bot can't help."""
    import requests
    telegram_token = os.environ.get("TELEGRAM_BOT_TOKEN")
    manager_chat = os.environ.get("MANAGER_TELEGRAM_CHAT_ID")

    if not telegram_token or not manager_chat:
        return

    message = (
        f"🔔 *Escalation from WhatsApp*\n\n"
        f"Customer: `{customer_number}`\n"
        f"Question: {question}\n\n"
        f"Bot replied: {bot_reply[:200]}..."
    )
    requests.post(
        f"https://api.telegram.org/bot{telegram_token}/sendMessage",
        json={
            "chat_id": manager_chat,
            "text": message,
            "parse_mode": "Markdown"
        }
    )


if __name__ == "__main__":
    app.run(port=5001, debug=False)

import requests
import json
import os

TEAMS_WEBHOOK_URL = os.environ["TEAMS_WEBHOOK_URL"]


def send_teams_notification(title: str, message: str,
                             color: str = "0078D4",
                             facts: list[dict] = None) -> bool:
    """
    Sends a notification card to a Teams channel through the webhook address from Workflows.
    Workflows accepts both MessageCard and Adaptive Card: check the format in a test channel.
    color: hex color of the side stripe (0078D4 = Microsoft blue)
    facts: a list of {"name": "...", "value": "..."} for a data table
    """
    card = {
        "@type": "MessageCard",
        "@context": "http://schema.org/extensions",
        "themeColor": color,
        "summary": title,
        "sections": [{
            "activityTitle": title,
            "activityText": message,
        }]
    }

    if facts:
        card["sections"][0]["facts"] = facts

    response = requests.post(
        TEAMS_WEBHOOK_URL,
        headers={"Content-Type": "application/json"},
        data=json.dumps(card)
    )
    return response.ok   # any successful response (200, 202)


# Example: a new deal notification
def notify_deal_closed(client_name: str, amount: float,
                        property_address: str) -> None:
    """A notification to the sales channel in Teams when a deal closes."""
    send_teams_notification(
        title="🎉 New deal closed!",
        message=f"An agent signed a contract with a client.",
        color="00B050",   # green
        facts=[
            {"name": "Client", "value": client_name},
            {"name": "Amount", "value": f"${amount:,.0f}"},
            {"name": "Property", "value": property_address},
        ]
    )

from flask import Flask, request, jsonify
import anthropic
import json
import os

app = Flask(__name__)
claude = anthropic.Anthropic()

ROUTING_SYSTEM = """You are a router for customer messages. Analyze the request and return JSON:
{
  "intent": "faq|lead|complaint|consultation|other",
  "urgency": "low|medium|high",
  "language": "en|es|ru",
  "should_escalate": true|false,
  "suggested_response": "a short answer if intent=faq"
}"""


def route_message(message: str, channel: str, user_id: str) -> dict:
    """
    Analyzes an incoming message and decides where it goes.
    channel: telegram|whatsapp|slack|teams
    """
    response = claude.messages.create(
        model="claude-haiku-4-5",   # Haiku for routing: fast and cheap; check that the model is still available
        max_tokens=256,
        system=ROUTING_SYSTEM,
        messages=[{
            "role": "user",
            "content": f"Channel: {channel}\nMessage: {message}"
        }]
    )

    try:
        return json.loads("".join(b.text for b in response.content if b.type == "text"))
    except json.JSONDecodeError:
        return {
            "intent": "other",
            "urgency": "medium",
            "language": "en",
            "should_escalate": False,
            "suggested_response": ""
        }


@app.route("/hub/incoming", methods=["POST"])
def unified_incoming():
    """
    A single entry point for messages from all channels.
    Each source (Telegram bot, WhatsApp webhook, Slack event)
    sends its message here in a standard format.
    """
    data = request.json
    channel = data.get("channel", "unknown")
    user_id = data.get("user_id", "")
    message = data.get("message", "")

    # 1. Routing with Claude Haiku (cheap and fast)
    routing = route_message(message, channel, user_id)

    # 2. Handling logic by intent
    if routing["should_escalate"]:
        # Hand off to a live person
        escalate_to_manager(channel, user_id, message, routing)
        reply = "I'm passing your question to a specialist. We'll reply within 1 hour."

    elif routing["intent"] == "faq" and routing["suggested_response"]:
        # Use the ready answer from routing
        reply = routing["suggested_response"]

    else:
        # A full answer from Claude Sonnet
        response = claude.messages.create(
            model="claude-sonnet-5-5",
            max_tokens=512,
            system="You are a friendly assistant. Reply in the customer's language.",
            messages=[{"role": "user", "content": message}]
        )
        reply = "".join(b.text for b in response.content if b.type == "text")

    return jsonify({
        "reply": reply,
        "routing": routing,
        "channel": channel
    })


def escalate_to_manager(channel: str, user_id: str,
                         message: str, routing: dict) -> None:
    """Notifies a manager on Telegram."""
    bot_token = os.environ.get("TELEGRAM_BOT_TOKEN")
    manager_chat = os.environ.get("MANAGER_TELEGRAM_CHAT_ID")
    if not bot_token or not manager_chat:
        return

    import requests
    requests.post(
        f"https://api.telegram.org/bot{bot_token}/sendMessage",
        json={
            "chat_id": manager_chat,
            "text": (
                f"🔔 *Escalation*\n"
                f"Channel: {channel} | User: `{user_id}`\n"
                f"Urgency: {routing.get('urgency', '?')}\n"
                f"Message: {message[:300]}"
            ),
            "parse_mode": "Markdown"
        }
    )


if __name__ == "__main__":
    app.run(port=5002, debug=False)

mkdir ai-messenger-bot && cd ai-messenger-bot
python -m venv venv && source venv/bin/activate
pip install python-telegram-bot anthropic python-dotenv flask requests

TELEGRAM_BOT_TOKEN=your_token
ANTHROPIC_API_KEY=sk-ant-...
TEAMS_WEBHOOK_URL=https://...  # we'll add this in step 3
MANAGER_TELEGRAM_CHAT_ID=your_chat_id

python bot.py

from dotenv import load_dotenv
load_dotenv()
notify_deal_closed("John Miller", 85000, "45 Calle Larga, Cuenca")