OpenCLAW + Discord + Telegram = Unstoppable Bot

Summary (what you’ll build)

•  A Python service that:
  • Receives messages from Discord and Telegram (bot accounts).
  • Forwards user messages to OpenCLAW via its API or local server.
  • Sends OpenCLAW’s responses back to the user on the same platform.
• Single process handles both platforms; design supports scaling with webhooks/workers.

Prerequisites

• Python 3.10+
• pip, virtualenv
• Discord bot token (create at Discord Developer Portal)
• Telegram bot token (via BotFather)
• OpenCLAW running as a service or accessible API endpoint (local or remote). You need the OpenCLAW API URL and API key (if applicable). If you don’t yet have OpenCLAW running, install/launch it per its docs (usually a server exposing REST/WS endpoints).
• Optional: ngrok (for local webhook testing) or a public host (VPS, Railway, Fly.io).

Architecture overview

• Bots receive messages via:
  • Discord: Gateway (discord.py) or webhooks (less common for interactive bots).
  • Telegram: Long polling (python-telegram-bot) or webhook.
• On message:
  • Normalize data (user id, chat id, text).
  • Call OpenCLAW API with a prompt payload (include any conversation context).
  • Receive response text (and attachments if supported).
  • Send response back via Discord/Telegram API.
• Persist conversation state (optional) in Redis or DB to support context/memory.

Libraries used (Python)

• discord.py (nextcord/py-cord forks also possible)
• python-telegram-bot (v20+)
• httpx or requests (for OpenCLAW HTTP calls)
• asyncio, aiohttp if needed
• optional: redis

Step‑by‑step: Project setup

1. Create project and virtualenv
   $ python -m venv venv
   $ source venv/bin/activate
   (venv) $ pip install discord.py python-telegram-bot httpx asyncio

2. Configuration
   Create a .env file (use python-dotenv or environment variables) with:
   DISCORD_TOKEN=your_discord_bot_token
   TELEGRAM_TOKEN=your_telegram_bot_token
   OPENCLAW_URL=http://localhost:8000/api/v1/respond # example
   OPENCLAW_API_KEY=your_openclaw_api_key_or_empty

3. Minimal code (single-file example)
   Save as bot_service.py (trimmed for clarity — replace OPENCLAW_URL with your endpoint):

import os
import asyncio
import logging
from dotenv import load_dotenv
import httpx

from discord import Intents
from discord.ext import commands
from telegram import Update, Bot
from telegram.ext import ApplicationBuilder, ContextTypes, MessageHandler, filters

load_dotenv()
DISCORD_TOKEN = os.getenv("DISCORD_TOKEN")
TELEGRAM_TOKEN = os.getenv("TELEGRAM_TOKEN")
OPENCLAW_URL = os.getenv("OPENCLAW_URL")
OPENCLAW_API_KEY = os.getenv("OPENCLAW_API_KEY", "")

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(name)

--- OpenCLAW client ---

async def call_openclaw(prompt: str, conversation_id: str = None) -> str:
payload = {
"prompt": prompt,
}
if conversation_id:
payload["conversation_id"] = conversation_id
headers = {}
if OPENCLAW_API_KEY:
headers["Authorization"] = f"Bearer {OPENCLAW_API_KEY}"
async with httpx.AsyncClient(timeout=30) as client:
resp = await client.post(OPENCLAW_URL, json=payload, headers=headers)
resp.raise_for_status()
data = resp.json()
# adapt to OpenCLAW response schema; here we expect {"response": "...", "conversation_id": "..."}
return data.get("response", ""), data.get("conversation_id")

--- Discord bot setup ---

intents = Intents.default()
intents.message_content = True
discord_bot = commands.Bot(command_prefix="!", intents=intents)

@discord_bot.event
async def on_ready():
logger.info(f"Discord bot logged in as {discord_bot.user}")

@discord_bot.event
async def on_message(message):
# ignore bot messages
if message.author.bot:
return
text = message.content
user = message.author
channel = message.channel

python

# optional: build prompt with context or system instructions
prompt = f"User ({user.name}): {text}\nAssistant:"
try:
    response_text, conv_id = await call_openclaw(prompt)
    await channel.send(response_text)
except Exception as e:
    logger.exception("OpenCLAW call failed")
    await channel.send("Sorry, I couldn't process that right now.")

--- Telegram bot setup ---

async def telegram_message_handler(update: Update, context: ContextTypes.DEFAULT_TYPE):
if not update.message or not update.message.text:
return
user = update.effective_user
chat_id = update.effective_chat.id
text = update.message.text

reasonml

prompt = f"User ({user.username or user.first_name}): {text}\nAssistant:"
try:
    response_text, conv_id = await call_openclaw(prompt)
    await context.bot.send_message(chat_id=chat_id, text=response_text)
except Exception:
    logger.exception("OpenCLAW call failed")
    await context.bot.send_message(chat_id=chat_id, text="Sorry, I couldn't process that right now.")

async def run_telegram():
app = ApplicationBuilder().token(TELEGRAM_TOKEN).build()
app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, telegram_message_handler))
await app.initialize()
await app.start()
# keep running until cancelled
await app.updater.start_polling() # for older versions; for v20+, use app.run_polling() in main thread

async def run_discord():
await discord_bot.start(DISCORD_TOKEN)

async def main():
# run Telegram and Discord concurrently
await asyncio.gather(run_telegram(), run_discord())

if name == "main":
try:
asyncio.run(main())
except KeyboardInterrupt:
logger.info("Shutting down")

Notes about the code

• Adjust OpenCLAW POST schema to match your server's API. Some OpenCLAW setups use websocket/streaming — adapt accordingly.
• For python-telegram-bot v20+, prefer Application.run_polling(); in an async context, use Application.start() and run the loop as above.
• For production, separate concerns: use worker queue for OpenCLAW calls, rate limit user requests, and handle concurrency.

Context and Memory

• To keep conversation context, pass a conversation_id from OpenCLAW (or maintain per-chat history in Redis/DB and include recent messages in the prompt).
• Store mapping: {platform_chat_id -> conversation_id}.

Advanced: Streaming and Rich Responses

• If OpenCLAW supports streaming responses, use websockets or server‑sent events and stream partial replies to Discord (edit messages) and Telegram (send_chat_action + edit_message_text).
• For attachments (images/files), have OpenCLAW return URLs or base64; upload via platform API accordingly.

Deployment

• Local testing: ngrok to expose Telegram webhooks if you want webhooks; otherwise long polling works for Telegram and Discord uses gateway (no public URL needed).
• Recommended hosts: VPS, Railway, Fly.io, Render, or Docker container on cloud VM. Use process manager (systemd, pm2 for node, or Docker + restart policy).
• Scale: Use a queue (RabbitMQ/Redis) and worker pool for OpenCLAW calls to handle bursts.

Security & Best Practices

• Keep tokens and API keys in environment variables; never commit to repo.
• Rate-limit per-user and per-channel to avoid abuse and high costs.
• Sanitize user input if you pass it to external services.
• Monitor logs and error rates; alert on OpenCLAW failures.
• Respect privacy: inform users that messages may be processed by an AI and log retention policies.
• For sensitive data, avoid sending PII to external endpoints or ensure your OpenCLAW instance is self-hosted and secure.

Troubleshooting

• “discord.py no attribute message_content” — enable message_content intent in Discord Developer Portal and in code intents.message_content = True.
• Telegram long polling vs webhooks: use polling for simple deployments; webhooks need a public HTTPS URL.
• OpenCLAW endpoint errors — test with curl/postman first, confirm request/response schema.

Optional: Example with conversation state (Redis)

• On each incoming message, look up conversation_id = redis.get(f"conv:{platform}:{chat_id}"), pass it to OpenCLAW, save returned conv_id back to Redis with TTL.