Skip to main content

​
Руководство по плагинам каналов

Создайте собственный канал для Велеса за три шага: создание подкласса, упаковка, установка.

​
Как это работает

Велес обнаруживает плагины каналов через точки входа Python. При запуске veles gateway сканируются:
  1. Встроенные каналы в veles/channels/
  2. Внешние пакеты, зарегистрированные в группе entry point veles.channels
Если соответствующий раздел конфигурации имеет значение "enabled": true, канал инициализируется и запускается.

​
Быстрый старт

Мы создадим минимальный канал веб-хука, который получает сообщения через HTTP POST и отправляет ответы обратно.

​
Структура проекта

veles-channel-webhook/
├── veles_channel_webhook/
│   ├── __init__.py          # реэкспорт WebhookChannel
│   └── channel.py           # реализация канала
└── pyproject.toml

​
1. Создание канала

# veles_channel_webhook/__init__.py
from veles_channel_webhook.channel import WebhookChannel

__all__ = ["WebhookChannel"]

# veles_channel_webhook/channel.py
import asyncio
from typing import Any

from aiohttp import web
from loguru import logger

from veles.channels.base import BaseChannel
from veles.bus.events import OutboundMessage


class WebhookChannel(BaseChannel):
    name = "webhook"
    display_name = "Webhook"

    @classmethod
    def default_config(cls) -> dict[str, Any]:
        return {"enabled": False, "port": 9000, "allowFrom": []}

    async def start(self) -> None:
        """Запуск HTTP-сервера для прослушивания входящих сообщений.

        ВАЖНО: start() должен блокировать выполнение (или пока не будет вызван stop()).
        Если функция завершается, канал считается неактивным.
        """
        self._running = True
        port = self.config.get("port", 9000)

        app = web.Application()
        app.router.add_post("/message", self._on_request)
        runner = web.AppRunner(app)
        await runner.setup()
        site = web.TCPSite(runner, "0.0.0.0", port)
        await site.start()
        logger.info("Webhook listening on :{}", port)

        # Блокировка до остановки
        while self._running:
            await asyncio.sleep(1)

        await runner.cleanup()

    async def stop(self) -> None:
        self._running = False

    async def send(self, msg: OutboundMessage) -> None:
        """Доставка исходящего сообщения.

        msg.content  — текст в формате markdown
        msg.media    — список путей к локальным файлам
        msg.chat_id  — получатель
        msg.metadata — может содержать "_progress": True для стриминга
        """
        logger.info("[webhook] -> {}: {}", msg.chat_id, msg.content[:80])

    async def _on_request(self, request: web.Request) -> web.Response:
        """Обработка входящего HTTP POST запроса."""
        body = await request.json()
        sender = body.get("sender", "unknown")
        chat_id = body.get("chat_id", sender)
        text = body.get("text", "")
        media = body.get("media", [])       # список URL

        # Основной вызов: валидация и отправка сообщения в шину для обработки агентом.
        await self._handle_message(
            sender_id=sender,
            chat_id=chat_id,
            content=text,
            media=media,
        )

        return web.json_response({"ok": True})

​
2. Регистрация Entry Point

# pyproject.toml
[project]
name = "veles-channel-webhook"
version = "0.1.0"
dependencies = ["veles", "aiohttp"]

[project.entry-points."veles.channels"]
webhook = "veles_channel_webhook:WebhookChannel"

[build-system]
requires = ["setuptools"]
build-backend = "setuptools.backends._legacy:_Backend"
Ключ (webhook) становится именем раздела конфигурации. Значение указывает на ваш подкласс BaseChannel.

​
3. Установка и настройка

pip install -e .
Veles plugins list      # проверка появления "Webhook" в списке
Veles onboard           # автодобавление конфига для обнаруженных плагинов
Отредактируйте ~/.veles/config.json: 
{
  "channels": {
    "webhook": {
      "enabled": true,
      "port": 9000,
      "allowFrom": ["*"]
    }
  }
}

​
4. Запуск и тестирование

Veles gateway
В другом терминале: 
curl -X POST http://localhost:9000/message \
  -H "Content-Type: application/json" \
  -d '{"sender": "user1", "chat_id": "user1", "text": "Привет!"}'
Агент получит сообщение и обработает его. Ответы придут в ваш метод send().

​
API BaseChannel

​
Обязательные методы (абстрактные)

МетодОписание
async start()Должен блокировать выполнение. Подключение к платформе, прослушивание сообщений, вызов _handle_message().
async stop()Установка self._running = False и очистка ресурсов. Вызывается при остановке gateway.
async send(msg: OutboundMessage)Доставка сообщения на платформу.

​
Базовые возможности

Метод / СвойствоОписание
_handle_message(sender_id, chat_id, content, media?, metadata?, session_key?)Вызывайте при получении сообщения. Проверяет доступ и публикует в шину.
is_allowed(sender_id)Проверка по списку config["allowFrom"].
default_config() (classmethod)Возвращает дефолтный конфиг для Veles onboard.
transcribe_audio(file_path)Транскрибация аудио через Groq Whisper (если настроено).
is_runningВозвращает состояние self._running.

​
Типы сообщений

@dataclass
class OutboundMessage:
    channel: str        # имя вашего канала
    chat_id: str        # ID получателя
    content: str        # текст сообщения (markdown)
    media: list[str]    # пути к локальным файлам (изображения, аудио и т.д.)
    metadata: dict      # метаданные: прогресс, ID сообщения для тредов и т.д.

​
Конфигурация

Ваш канал получает конфиг как обычный dict. Доступ к полям через .get(): 
async def start(self) -> None:
    port = self.config.get("port", 9000)
    token = self.config.get("token", "")
allowFrom обрабатывается автоматически в _handle_message(). Переопределите default_config(), чтобы Veles onboard автоматически заполнил config.json: 
@classmethod
def default_config(cls) -> dict[str, Any]:
    return {"enabled": False, "port": 9000, "allowFrom": []}

​
Соглашение об именовании

ОбъектФорматПример
Пакет PyPIveles-channel-{name}veles-channel-webhook
Ключ Entry point{name}webhook
Раздел конфигаchannels.{name}channels.webhook
Пакет Pythonveles_channel_{name}veles_channel_webhook

​
Локальная разработка

git clone https://github.com/you/veles-channel-webhook
cd veles-channel-webhook
pip install -e .
Veles plugins list    # должен показать "Webhook"
Veles gateway         # полное тестирование

​
Проверка

$ Veles plugins list

  Name       Source    Enabled
  telegram   builtin  yes
  discord    builtin  no
  webhook    plugin   yes