Telegram Bot

7123456789:AAHxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx

pip install aiogram aiohttp

BOT_TOKEN=ваш_токен_от_BotFather
API_KEY=YOUR_API_KEY
API_URL=https://api.claudecodeapi.cloud

import os
import json
import asyncio
import aiohttp
from aiogram import Bot, Dispatcher, types
from aiogram.filters import Command
from collections import defaultdict

BOT_TOKEN = os.getenv("BOT_TOKEN")
API_KEY = os.getenv("API_KEY")
API_URL = os.getenv("API_URL", "https://api.claudecodeapi.cloud")

bot = Bot(token=BOT_TOKEN)
dp = Dispatcher()

# История диалогов для каждого пользователя
user_histories = defaultdict(list)
MAX_HISTORY = 20


async def ask_claude(user_id: int, text: str) -> str:
    """Отправка запроса к Claude API"""
    user_histories[user_id].append({
        "role": "user",
        "content": text
    })

    # Обрезаем историю если слишком длинная
    if len(user_histories[user_id]) > MAX_HISTORY:
        user_histories[user_id] = user_histories[user_id][-MAX_HISTORY:]

    headers = {
        "x-api-key": API_KEY,
        "content-type": "application/json",
        "anthropic-version": "2023-06-01"
    }

    payload = {
        "model": "claude-sonnet-4-6",
        "max_tokens": 4096,
        "messages": user_histories[user_id]
    }

    async with aiohttp.ClientSession() as session:
        async with session.post(
            f"{API_URL}/v1/messages",
            headers=headers,
            json=payload
        ) as resp:
            if resp.status != 200:
                error = await resp.text()
                raise Exception(f"API error {resp.status}: {error}")

            data = await resp.json()
            reply = data["content"][0]["text"]

            user_histories[user_id].append({
                "role": "assistant",
                "content": reply
            })

            return reply


@dp.message(Command("start"))
async def cmd_start(message: types.Message):
    await message.answer(
        "Привет! Я AI-ассистент на базе Claude.\n\n"
        "Просто напишите сообщение — я отвечу.\n"
        "/clear — очистить историю диалога"
    )


@dp.message(Command("clear"))
async def cmd_clear(message: types.Message):
    user_histories[message.from_user.id].clear()
    await message.answer("История диалога очищена.")


@dp.message()
async def handle_message(message: types.Message):
    if not message.text:
        return

    thinking = await message.answer("Думаю...")

    try:
        reply = await ask_claude(message.from_user.id, message.text)

        # Telegram лимит — 4096 символов
        if len(reply) <= 4096:
            await thinking.edit_text(reply, parse_mode="Markdown")
        else:
            await thinking.delete()
            for i in range(0, len(reply), 4096):
                await message.answer(reply[i:i+4096])

    except Exception as e:
        await thinking.edit_text(f"Ошибка: {e}")


async def main():
    print("Бот запущен!")
    await dp.start_polling(bot)


if __name__ == "__main__":
    from dotenv import load_dotenv
    load_dotenv()
    asyncio.run(main())

python bot.py

[Unit]
Description=Claude Telegram Bot
After=network.target

[Service]
Type=simple
User=root
WorkingDirectory=/root/claude-bot
EnvironmentFile=/root/claude-bot/.env
ExecStart=/usr/bin/python3 /root/claude-bot/bot.py
Restart=always
RestartSec=10

[Install]
WantedBy=multi-user.target

sudo systemctl enable claude-bot
sudo systemctl start claude-bot
sudo systemctl status claude-bot