pybotx (expressapp/pybotx)

pybotx

https://github.com/expressapp/pybotx
Admin
A library for creating chatbots and SmartApps for the eXpress messenger, designed to be easy to use...

Tokens:15,356
Snippets:76
Trust Score:6.2
Update:1 month ago
Show doc for...
Context Summary (auto-generated)
Raw
# pybotx

Библиотека pybotx — это Python-решение для создания чат-ботов и SmartApps для корпоративного мессенджера eXpress. Она предоставляет полноценный SDK для взаимодействия с платформой BotX, поддерживает асинхронную обработку команд, систему middleware, коллбэки и типизацию через аннотации. Библиотека легко интегрируется с современными асинхронными веб-фреймворками, такими как FastAPI.

Основная функциональность включает: обработку входящих сообщений и системных событий, отправку и редактирование сообщений, работу с файлами, управление чатами, поиск пользователей, SmartApp-события, стикеры и метрики. pybotx использует Pydantic для валидации данных и httpx для асинхронных HTTP-запросов, обеспечивая надёжную и производительную работу бота.

## Инициализация бота (Bot)

Класс `Bot` является центральным компонентом библиотеки. Он управляет обработчиками команд, подключениями к BotX API и жизненным циклом бота. При инициализации передаются коллекторы обработчиков, учётные данные бота, middleware и обработчики исключений.

```python
from uuid import UUID
from pybotx import Bot, BotAccountWithSecret, HandlerCollector

collector = HandlerCollector()

@collector.command("/hello", description="Приветствие пользователя")
async def hello_handler(message, bot):
    await bot.answer_message(f"Привет, {message.sender.username}!")

bot = Bot(
    collectors=[collector],
    bot_accounts=[
        BotAccountWithSecret(
            id=UUID("123e4567-e89b-12d3-a456-426655440000"),
            cts_url="https://cts.example.com",
            secret_key="your_secret_key_here",
        ),
    ],
    exception_handlers={Exception: lambda msg, bot, exc: bot.answer_message("Ошибка!")},
)

# Интеграция с FastAPI
from fastapi import FastAPI
app = FastAPI()
app.add_event_handler("startup", bot.startup)
app.add_event_handler("shutdown", bot.shutdown)
```

## Регистрация обработчиков команд (@collector.command)

Декоратор `@collector.command` регистрирует функцию-обработчик для конкретной команды бота. Команды начинаются с `/`, могут быть видимыми или скрытыми, и поддерживают динамическую видимость через callback-функции.

```python
from uuid import UUID
from pybotx import HandlerCollector, IncomingMessage, Bot, StatusRecipient

ADMIN_HUIDS = {UUID("123e4567-e89b-12d3-a456-426614174000")}
collector = HandlerCollector()

# Видимая команда (отображается в меню бота)
@collector.command("/help", description="Показать справку")
async def help_handler(message: IncomingMessage, bot: Bot) -> None:
    await bot.answer_message("Доступные команды: /help, /echo, /admin")

# Скрытая команда (не отображается в меню)
@collector.command("/_debug", visible=False)
async def debug_handler(message: IncomingMessage, bot: Bot) -> None:
    await bot.answer_message(f"Debug info: {message.raw_command}")

# Команда с динамической видимостью (только для админов)
async def is_admin(status_recipient: StatusRecipient, bot: Bot) -> bool:
    return status_recipient.huid in ADMIN_HUIDS

@collector.command("/admin", visible=is_admin, description="Админ-панель")
async def admin_handler(message: IncomingMessage, bot: Bot) -> None:
    await bot.answer_message("Добро пожаловать в админ-панель!")

# Обработчик по умолчанию (для неизвестных команд)
@collector.default_message_handler
async def default_handler(message: IncomingMessage, bot: Bot) -> None:
    await bot.answer_message(f"Неизвестная команда: {message.body}")
```

## Отправка сообщений (answer_message, send_message)

Методы `answer_message` и `send_message` позволяют отправлять сообщения пользователям. `answer_message` автоматически определяет получателя из контекста входящего сообщения, а `send_message` требует явного указания `chat_id`.

```python
from uuid import UUID
from pybotx import IncomingMessage, Bot, HandlerCollector, OutgoingMessage

collector = HandlerCollector()

@collector.command("/echo", description="Эхо-ответ")
async def echo_handler(message: IncomingMessage, bot: Bot) -> None:
    # Простой ответ в текущий чат
    await bot.answer_message(f"Вы написали: {message.body}")

@collector.command("/notify", description="Отправить уведомление")
async def notify_handler(message: IncomingMessage, bot: Bot) -> None:
    target_chat_id = UUID(message.argument)  # UUID чата из аргумента команды

    # Отправка в другой чат
    sync_id = await bot.send_message(
        bot_id=message.bot.id,
        chat_id=target_chat_id,
        body="Уведомление от бота!",
        silent_response=False,  # Показать в истории
        send_push=True,         # Отправить push-уведомление
        ignore_mute=False,      # Учитывать настройки "Не беспокоить"
    )
    await bot.answer_message(f"Сообщение отправлено, sync_id: {sync_id}")

@collector.command("/prebuild", description="Отправка готового сообщения")
async def prebuild_handler(message: IncomingMessage, bot: Bot) -> None:
    # Использование OutgoingMessage для формирования сообщения
    outgoing = OutgoingMessage(
        bot_id=message.bot.id,
        chat_id=message.chat.id,
        body="Сформированное сообщение",
        metadata={"custom_field": "value"},
    )
    await bot.send(message=outgoing)
```

## Сообщения с кнопками (BubbleMarkup, KeyboardMarkup)

Библиотека поддерживает интерактивные кнопки двух типов: `BubbleMarkup` (кнопки под сообщением) и `KeyboardMarkup` (кнопки под полем ввода). Кнопки могут выполнять команды бота или открывать внешние ссылки.

```python
from pybotx import IncomingMessage, Bot, HandlerCollector, BubbleMarkup, KeyboardMarkup

collector = HandlerCollector()

@collector.command("/menu", description="Показать меню")
async def menu_handler(message: IncomingMessage, bot: Bot) -> None:
    bubbles = BubbleMarkup()

    # Кнопки с командами в одной строке
    bubbles.add_button(
        command="/action",
        label="Действие 1",
        data={"action": "first"},
        background_color="#4CAF50",
    )
    bubbles.add_button(
        command="/action",
        label="Действие 2",
        data={"action": "second"},
        background_color="#2196F3",
        new_row=False,  # Та же строка
    )

    # Кнопка-ссылка
    bubbles.add_button(
        label="Открыть сайт",
        link="https://express.ms",
        alert="Вы будете перенаправлены на внешний сайт",
    )

    await bot.answer_message("Выберите действие:", bubbles=bubbles)

@collector.command("/action", visible=False)
async def action_handler(message: IncomingMessage, bot: Bot) -> None:
    action = message.data.get("action", "unknown")
    await bot.answer_message(f"Вы выбрали: {action}")

@collector.command("/keyboard", description="Показать клавиатуру")
async def keyboard_handler(message: IncomingMessage, bot: Bot) -> None:
    keyboard = KeyboardMarkup()
    keyboard.add_button(command="/yes", label="Да")
    keyboard.add_button(command="/no", label="Нет", new_row=False)

    await bot.answer_message("Вы согласны?", keyboard=keyboard)
```

## Редактирование и удаление сообщений (edit_message, delete_message)

Бот может редактировать ранее отправленные сообщения и удалять их. Для этого используется `sync_id` — уникальный идентификатор сообщения.

```python
from pybotx import IncomingMessage, Bot, HandlerCollector, BubbleMarkup

collector = HandlerCollector()

@collector.command("/counter", description="Интерактивный счётчик")
async def counter_handler(message: IncomingMessage, bot: Bot) -> None:
    # source_sync_id указывает на сообщение, в котором нажата кнопка
    if message.source_sync_id:
        current = message.data.get("value", 0)
        new_value = current + 1

        bubbles = BubbleMarkup()
        bubbles.add_button(command="/counter", label="+1", data={"value": new_value})
        bubbles.add_button(command="/delete", label="Удалить", data={"sync_id": str(message.source_sync_id)}, new_row=False)

        # Редактирование существующего сообщения
        await bot.edit_message(
            bot_id=message.bot.id,
            sync_id=message.source_sync_id,
            body=f"Счётчик: {new_value}",
            bubbles=bubbles,
        )
    else:
        bubbles = BubbleMarkup()
        bubbles.add_button(command="/counter", label="+1", data={"value": 1})

        await bot.answer_message("Счётчик: 0", bubbles=bubbles)

@collector.command("/delete", visible=False)
async def delete_handler(message: IncomingMessage, bot: Bot) -> None:
    from uuid import UUID
    sync_id = UUID(message.data["sync_id"])

    # Удаление сообщения
    await bot.delete_message(bot_id=message.bot.id, sync_id=sync_id)
```

## Работа с файлами (OutgoingAttachment, download_file, upload_file)

Библиотека поддерживает отправку и получение файлов. Файлы могут быть созданы из буфера в памяти или загружены с файловой системы.

```python
from aiofiles.tempfile import NamedTemporaryFile
from pybotx import IncomingMessage, Bot, HandlerCollector, OutgoingAttachment

collector = HandlerCollector()

@collector.command("/send-file", description="Отправить файл")
async def send_file_handler(message: IncomingMessage, bot: Bot) -> None:
    async with NamedTemporaryFile("wb+") as buffer:
        await buffer.write(b"Hello, World!\nThis is a test file.")
        await buffer.seek(0)

        attachment = await OutgoingAttachment.from_async_buffer(buffer, "hello.txt")

    await bot.answer_message("Вот ваш файл:", file=attachment)

@collector.command("/echo-file", description="Вернуть полученный файл")
async def echo_file_handler(message: IncomingMessage, bot: Bot) -> None:
    if not message.file:
        await bot.answer_message("Пожалуйста, прикрепите файл к сообщению")
        return

    # message.file уже содержит IncomingFileAttachment
    await bot.answer_message(
        f"Получен файл: {message.file.filename} ({message.file.size} байт)",
        file=message.file,
    )

@collector.command("/download", description="Скачать файл с сервера")
async def download_handler(message: IncomingMessage, bot: Bot) -> None:
    from uuid import UUID

    file_id = UUID(message.argument)

    async with NamedTemporaryFile("wb+") as buffer:
        await bot.download_file(
            bot_id=message.bot.id,
            chat_id=message.chat.id,
            file_id=file_id,
            async_buffer=buffer,
        )
        await buffer.seek(0)
        content = await buffer.read()

    await bot.answer_message(f"Файл скачан, размер: {len(content)} байт")
```

## Упоминания пользователей (MentionBuilder)

`MentionBuilder` позволяет создавать упоминания пользователей, чатов и каналов в сообщениях бота.

```python
from pybotx import IncomingMessage, Bot, HandlerCollector, MentionBuilder

collector = HandlerCollector()

@collector.command("/mention", description="Упомянуть пользователя")
async def mention_handler(message: IncomingMessage, bot: Bot) -> None:
    # Упоминание отправителя
    sender_mention = MentionBuilder.contact(message.sender.huid)
    await bot.answer_message(f"Привет, {sender_mention}!")

@collector.command("/mention-all", description="Упомянуть всех")
async def mention_all_handler(message: IncomingMessage, bot: Bot) -> None:
    all_mention = MentionBuilder.all()
    await bot.answer_message(f"{all_mention} Внимание всем!")

@collector.command("/list-mentions", description="Показать упоминания из сообщения")
async def list_mentions_handler(message: IncomingMessage, bot: Bot) -> None:
    if not message.mentions:
        await bot.answer_message("В сообщении нет упоминаний")
        return

    mentions_info = []
    for mention in message.mentions.users:
        mentions_info.append(f"Пользователь: {mention.entity_id}")
    for mention in message.mentions.contacts:
        mentions_info.append(f"Контакт: {mention.entity_id}")

    await bot.answer_message("Упоминания:\n" + "\n".join(mentions_info))
```

## Обработка системных событий

Библиотека поддерживает обработку различных системных событий: создание чата, добавление/удаление пользователей, SmartApp-события и другие.

```python
from pybotx import (
    HandlerCollector, Bot, ChatCreatedEvent, AddedToChatEvent,
    DeletedFromChatEvent, SmartAppEvent, InternalBotNotificationEvent,
)

collector = HandlerCollector()

@collector.chat_created
async def on_chat_created(event: ChatCreatedEvent, bot: Bot) -> None:
    await bot.send_message(
        bot_id=event.bot.id,
        chat_id=event.chat_id,
        body=f"Чат создан! Участники: {len(event.members)}",
    )

@collector.added_to_chat
async def on_added_to_chat(event: AddedToChatEvent, bot: Bot) -> None:
    for huid in event.huids:
        await bot.send_message(
            bot_id=event.bot.id,
            chat_id=event.chat_id,
            body=f"Добро пожаловать в чат!",
        )

@collector.deleted_from_chat
async def on_deleted_from_chat(event: DeletedFromChatEvent, bot: Bot) -> None:
    print(f"Пользователи {event.huids} удалены из чата {event.chat_id}")

@collector.smartapp_event
async def on_smartapp_event(event: SmartAppEvent, bot: Bot) -> None:
    print(f"SmartApp событие: {event.data}")
    await bot.send_smartapp_event(
        bot_id=event.bot.id,
        chat_id=event.chat.id,
        data={"status": "received", "ref": str(event.ref)},
    )

@collector.internal_bot_notification
async def on_internal_notification(event: InternalBotNotificationEvent, bot: Bot) -> None:
    print(f"Внутреннее уведомление от бота: {event.data}")
```

## Middleware

Middleware позволяют выполнять код до и после обработки команды. Они используются для логирования, проверки прав доступа, инъекции зависимостей и других сквозных задач.

```python
from httpx import AsyncClient
from pybotx import HandlerCollector, IncomingMessage, Bot, IncomingMessageHandlerFunc

collector = HandlerCollector()

async def logging_middleware(
    message: IncomingMessage,
    bot: Bot,
    call_next: IncomingMessageHandlerFunc,
) -> None:
    print(f"[LOG] Команда: {message.body} от {message.sender.huid}")
    await call_next(message, bot)
    print(f"[LOG] Обработка завершена")

async def admin_check_middleware(
    message: IncomingMessage,
    bot: Bot,
    call_next: IncomingMessageHandlerFunc,
) -> None:
    ADMIN_HUIDS = {"123e4567-e89b-12d3-a456-426614174000"}
    if str(message.sender.huid) not in ADMIN_HUIDS:
        await bot.answer_message("Доступ запрещён!")
        return
    await call_next(message, bot)

async def http_client_middleware(
    message: IncomingMessage,
    bot: Bot,
    call_next: IncomingMessageHandlerFunc,
) -> None:
    async with AsyncClient() as client:
        message.state.http_client = client
        await call_next(message, bot)

# Применение middleware к отдельной команде
@collector.command(
    "/admin-action",
    description="Админ-действие",
    middlewares=[admin_check_middleware],
)
async def admin_action(message: IncomingMessage, bot: Bot) -> None:
    await bot.answer_message("Админ-действие выполнено!")

# Создание коллектора с глобальными middleware
admin_collector = HandlerCollector(middlewares=[logging_middleware, admin_check_middleware])

@admin_collector.command("/secure", description="Защищённая команда")
async def secure_handler(message: IncomingMessage, bot: Bot) -> None:
    await bot.answer_message("Вы прошли проверку!")
```

## Управление чатами (create_chat, chat_info, add_users_to_chat)

Бот может создавать новые чаты, получать информацию о существующих чатах и управлять их участниками.

```python
from uuid import UUID
from pybotx import (
    IncomingMessage, Bot, HandlerCollector, ChatTypes, MentionBuilder,
    ChatCreationError, ChatCreationProhibitedError,
)

collector = HandlerCollector()

@collector.command("/create-group", description="Создать групповой чат")
async def create_group_handler(message: IncomingMessage, bot: Bot) -> None:
    contacts = message.mentions.contacts
    if not contacts:
        await bot.answer_message("Упомяните пользователей для добавления в чат")
        return

    try:
        chat_id = await bot.create_chat(
            bot_id=message.bot.id,
            name="Новый групповой чат",
            chat_type=ChatTypes.GROUP_CHAT,
            huids=[contact.entity_id for contact in contacts],
            description="Чат создан ботом",
            shared_history=True,
        )
        chat_mention = MentionBuilder.chat(chat_id)
        await bot.answer_message(f"Создан чат: {chat_mention}")
    except (ChatCreationError, ChatCreationProhibitedError) as e:
        await bot.answer_message(f"Ошибка создания чата: {e}")

@collector.command("/chat-info", description="Информация о чате")
async def chat_info_handler(message: IncomingMessage, bot: Bot) -> None:
    info = await bot.chat_info(bot_id=message.bot.id, chat_id=message.chat.id)

    members_count = len(info.members)
    admin_count = sum(1 for m in info.members if m.is_admin)

    await bot.answer_message(
        f"Название: {info.name}\n"
        f"Тип: {info.chat_type}\n"
        f"Участников: {members_count}\n"
        f"Админов: {admin_count}"
    )

@collector.command("/add-user", description="Добавить пользователя в чат")
async def add_user_handler(message: IncomingMessage, bot: Bot) -> None:
    user_huid = UUID(message.argument)

    await bot.add_users_to_chat(
        bot_id=message.bot.id,
        chat_id=message.chat.id,
        huids=[user_huid],
    )
    await bot.answer_message("Пользователь добавлен!")
```

## Поиск пользователей (search_user_by_huid, search_user_by_email)

Библиотека предоставляет методы для поиска пользователей по различным идентификаторам: HUID, email, AD-логин.

```python
import dataclasses
from pybotx import IncomingMessage, Bot, HandlerCollector, UserNotFoundError

collector = HandlerCollector()

@collector.command("/whoami", description="Информация о себе")
async def whoami_handler(message: IncomingMessage, bot: Bot) -> None:
    try:
        user = await bot.search_user_by_huid(
            bot_id=message.bot.id,
            huid=message.sender.huid,
        )
        info = dataclasses.asdict(user)
        await bot.answer_message(f"Ваш профиль:\n{info}")
    except UserNotFoundError:
        await bot.answer_message("Пользователь не найден")

@collector.command("/find-by-email", description="Найти пользователя по email")
async def find_by_email_handler(message: IncomingMessage, bot: Bot) -> None:
    email = message.argument
    if not email:
        await bot.answer_message("Укажите email: /find-by-email user@example.com")
        return

    try:
        user = await bot.search_user_by_email_post(
            bot_id=message.bot.id,
            email=email,
        )
        await bot.answer_message(
            f"Найден пользователь:\n"
            f"Имя: {user.name}\n"
            f"HUID: {user.huid}\n"
            f"Email: {user.email}"
        )
    except UserNotFoundError:
        await bot.answer_message(f"Пользователь с email {email} не найден")

@collector.command("/list-users", description="Список всех пользователей")
async def list_users_handler(message: IncomingMessage, bot: Bot) -> None:
    users_list = []
    async with bot.users_as_csv(
        bot_id=message.bot.id,
        cts_user=True,
        unregistered=False,
        botx=False,
    ) as users:
        async for user in users:
            users_list.append(f"{user.name} ({user.email})")
            if len(users_list) >= 10:
                break

    await bot.answer_message("Пользователи:\n" + "\n".join(users_list))
```

## SmartApp события (send_smartapp_event, sync_smartapp_event)

SmartApps — это интерактивные приложения внутри мессенджера. Библиотека поддерживает как асинхронные, так и синхронные SmartApp-события.

```python
from pybotx import (
    HandlerCollector, Bot, SmartAppEvent,
    BotAPISyncSmartAppEventResultResponse,
)

collector = HandlerCollector()

# Асинхронный обработчик SmartApp событий
@collector.smartapp_event
async def handle_smartapp_event(event: SmartAppEvent, bot: Bot) -> None:
    action = event.data.get("action")

    if action == "get_data":
        await bot.send_smartapp_event(
            bot_id=event.bot.id,
            chat_id=event.chat.id,
            ref=event.ref,
            data={"items": [1, 2, 3], "total": 3},
            encrypted=True,
        )
    elif action == "update":
        # Обработка обновления
        await bot.send_smartapp_notification(
            bot_id=event.bot.id,
            chat_id=event.chat.id,
            smartapp_counter=5,
            body="Данные обновлены",
        )

# Синхронный обработчик (для эндпоинта /smartapps/request)
@collector.sync_smartapp_event
async def handle_sync_event(
    event: SmartAppEvent,
    bot: Bot,
) -> BotAPISyncSmartAppEventResultResponse:
    result = {"processed": True, "input": event.data}

    return BotAPISyncSmartAppEventResultResponse.from_domain(
        data=result,
        files=[],
    )
```

## Интеграция с FastAPI (полный пример)

Полный пример интеграции pybotx с FastAPI, включающий все необходимые эндпоинты для работы бота.

```python
from http import HTTPStatus
from uuid import UUID

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

from pybotx import (
    Bot, BotAccountWithSecret, HandlerCollector, IncomingMessage,
    build_command_accepted_response, ChatCreatedEvent,
)

collector = HandlerCollector()

@collector.command("/ping", description="Проверка связи")
async def ping_handler(message: IncomingMessage, bot: Bot) -> None:
    await bot.answer_message("pong!")

@collector.chat_created
async def on_chat_created(event: ChatCreatedEvent, bot: Bot) -> None:
    await bot.send_message(
        bot_id=event.bot.id,
        chat_id=event.chat_id,
        body="Бот готов к работе!",
    )

bot = Bot(
    collectors=[collector],
    bot_accounts=[
        BotAccountWithSecret(
            id=UUID("123e4567-e89b-12d3-a456-426655440000"),
            cts_url="https://cts.example.com",
            secret_key="your_secret_key",
        ),
    ],
)

app = FastAPI(title="eXpress Bot")

@app.on_event("startup")
async def startup():
    await bot.startup()

@app.on_event("shutdown")
async def shutdown():
    await bot.shutdown()

@app.post("/command")
async def command_handler(request: Request) -> JSONResponse:
    bot.async_execute_raw_bot_command(
        await request.json(),
        request_headers=request.headers,
    )
    return JSONResponse(
        build_command_accepted_response(),
        status_code=HTTPStatus.ACCEPTED,
    )

@app.post("/smartapps/request")
async def smartapp_handler(request: Request) -> JSONResponse:
    response = await bot.sync_execute_raw_smartapp_event(
        await request.json(),
        request_headers=request.headers,
    )
    return JSONResponse(response.jsonable_dict(), status_code=HTTPStatus.OK)

@app.get("/status")
async def status_handler(request: Request) -> JSONResponse:
    status = await bot.raw_get_status(
        dict(request.query_params),
        request_headers=request.headers,
    )
    return JSONResponse(status)

@app.post("/notification/callback")
async def callback_handler(request: Request) -> JSONResponse:
    await bot.set_raw_botx_method_result(
        await request.json(),
        verify_request=False,
    )
    return JSONResponse(
        build_command_accepted_response(),
        status_code=HTTPStatus.ACCEPTED,
    )
```

## Обработка ошибок (exception_handlers)

Библиотека позволяет централизованно обрабатывать исключения, возникающие в обработчиках команд.

```python
from loguru import logger
from pybotx import Bot, BotAccountWithSecret, HandlerCollector, IncomingMessage

async def handle_value_error(
    message: IncomingMessage,
    bot: Bot,
    exc: ValueError,
) -> None:
    await bot.answer_message(f"Ошибка валидации: {exc}")

async def handle_any_error(
    message: IncomingMessage,
    bot: Bot,
    exc: Exception,
) -> None:
    logger.exception("Необработанная ошибка:")
    await bot.answer_message("Произошла внутренняя ошибка. Попробуйте позже.")

collector = HandlerCollector()

@collector.command("/divide", description="Деление чисел")
async def divide_handler(message: IncomingMessage, bot: Bot) -> None:
    args = message.arguments
    if len(args) != 2:
        raise ValueError("Нужно два числа: /divide 10 2")

    a, b = float(args[0]), float(args[1])
    result = a / b  # Может вызвать ZeroDivisionError
    await bot.answer_message(f"Результат: {result}")

bot = Bot(
    collectors=[collector],
    bot_accounts=[],
    exception_handlers={
        ValueError: handle_value_error,
        Exception: handle_any_error,
    },
)
```

## Стикеры (create_sticker_pack, add_sticker)

Библиотека поддерживает создание и управление стикерпаками.

```python
from uuid import UUID
from aiofiles.tempfile import NamedTemporaryFile
from pybotx import IncomingMessage, Bot, HandlerCollector

collector = HandlerCollector()

@collector.command("/create-stickers", description="Создать стикерпак")
async def create_stickers_handler(message: IncomingMessage, bot: Bot) -> None:
    # Создание пустого стикерпака
    sticker_pack = await bot.create_sticker_pack(
        bot_id=message.bot.id,
        name="Мои стикеры",
        huid=message.sender.huid,
    )

    await bot.answer_message(f"Создан стикерпак: {sticker_pack.name} (ID: {sticker_pack.id})")

@collector.command("/add-sticker", description="Добавить стикер")
async def add_sticker_handler(message: IncomingMessage, bot: Bot) -> None:
    if not message.file:
        await bot.answer_message("Прикрепите PNG-изображение")
        return

    pack_id = UUID(message.argument)

    # Стикер должен быть PNG-файлом
    sticker = await bot.add_sticker(
        bot_id=message.bot.id,
        sticker_pack_id=pack_id,
        emoji="😀",
        async_buffer=message.file,  # Буфер с PNG-изображением
    )

    await bot.answer_message(f"Стикер добавлен: {sticker.emoji}")

@collector.command("/my-stickers", description="Мои стикерпаки")
async def my_stickers_handler(message: IncomingMessage, bot: Bot) -> None:
    packs = []
    async for pack in bot.iterate_by_sticker_packs(
        bot_id=message.bot.id,
        user_huid=message.sender.huid,
    ):
        packs.append(f"- {pack.name} ({pack.stickers_count} стикеров)")

    if packs:
        await bot.answer_message("Ваши стикерпаки:\n" + "\n".join(packs))
    else:
        await bot.answer_message("У вас нет стикерпаков")
```

---

Библиотека pybotx идеально подходит для создания корпоративных ботов и SmartApps в экосистеме eXpress. Основные сценарии использования включают: автоматизацию рабочих процессов, интеграцию с внутренними системами компании, создание интерактивных меню и опросов, уведомления о событиях, а также разработку полноценных SmartApp-приложений с богатым пользовательским интерфейсом.

Библиотека следует современным практикам Python-разработки: использует asyncio для асинхронной обработки, Pydantic для валидации данных, полную типизацию и 100% покрытие тестами. Интеграция с веб-фреймворками (FastAPI, Starlette) осуществляется через простые эндпоинты, что позволяет легко встроить бота в существующую инфраструктуру. Система middleware и обработчиков исключений обеспечивает гибкость и надёжность при обработке запросов.