Основные компоненты¶

LXMFBot¶

Основной класс бота, который обрабатывает маршрутизацию сообщений, обработку команд и управление жизненным циклом бота.

from lxmfy import LXMFBot

bot = LXMFBot(
    name="MyBot",
    announce=600,
    announce_immediately=True,
    admins=set(),
    hot_reloading=False,
    rate_limit=5,
    cooldown=60,
    max_warnings=3,
    warning_timeout=300,
    command_prefix="/",
    cogs_dir="cogs",
    cogs_enabled=True,
    permissions_enabled=False,
    storage_type="json", # "json", "sqlite", or "memory"
    storage_path="data",
    first_message_enabled=True,
    event_logging_enabled=True,
    max_logged_events=1000,
    event_middleware_enabled=True,
    announce_enabled=True,
    signature_verification_enabled=False,
    require_message_signatures=False,
    identity_pinning_enabled=False,
    message_persistence_enabled=True,
    dynamic_cogs_enabled=True,
    external_cogs_enabled=True,
    external_cogs_sandbox_enabled=True,
    external_cogs_sandbox_type="auto",  # "auto", "landlock", "bwrap", "firejail", "none"
    external_cogs_timeout=30,
    landlock_enabled=True,
    nlp_enabled=False,
    nlp_threshold=0.5,
    link_support_enabled=False,
    lxmf_commands_enabled=True,
    message_queue_size=50,
    reticulum_config_dir=None,  # or LXMFY_RETICULUM_CONFIG_DIR / "~/.reticulum"
    rrc_enabled=False,
    rrc_hubs=[],
    rrc_rooms=[],
    rrc_nick=None,
    rrc_dest_name="rrc.hub",
    rrc_auto_reconnect=True,
    rrc_persist_sessions=True,
)

Ключевые методы¶

get_landlock_status(): Return Landlock LSM sandbox availability and activation state for the bot process
run(delay=10): Запустить основной цикл бота
send(destination, message, title="Reply", lxmf_fields=None, stamp_cost=None, opportunistic=None): Send a message to a destination, optionally with custom LXMF fields, stamp cost override, and opportunistic sending (tries direct, falls back to propagation immediately if configured).
send_with_attachment(destination, message, attachment, title="Reply", stamp_cost=None, opportunistic=None): Send a message with an attachment
command(name, description="No description provided", admin_only=False, threaded=False): Decorator for registering commands. Set threaded=True to run the command’s callback in a separate thread. Commands support type-hinted arguments for automatic conversion.
intent(name, examples): Decorator for registering NLP intent handlers.
nlp.export_model(): Export trained NLP model data.
nlp.import_model(model_data): Import previously exported NLP model data.
request_link(destination_hash, callback=None, app_name="lxmf", *aspects): Request an RNS link to a destination. Allows custom app_name and aspects (defaults to «lxmf» and «delivery»).
on_link(callback): Register a handler for incoming RNS links.
load_extension(name): Load a cog extension module by name (e.g., «cogs.utility»).
reload_extension(name): Reload a cog extension module.
add_cog(cog_instance): Add a cog class instance to the bot.
remove_cog(cog_name): Remove a cog from the bot by its class name.
on_first_message(): Декоратор для обработки первых сообщений от пользователей
on_message(): Decorator for handling all messages (called before command processing)
validate(): Запустить проверочные тесты конфигурации бота
connect_rrc(hub_hash, rooms=None, nick=None, dest_name=None, auto_reconnect=None): Connect to an RRC hub as a client
disconnect_rrc(hub_hash=None): Disconnect one or all RRC hub sessions
on_rrc(callback=None): Decorator or register handler for RRC events (handler(event, client, payload))
rrc: RRCManager instance for multi-hub sessions

Structured Commands via LXMF Fields¶

Bots can receive commands sent via LXMF FIELD_COMMANDS (0x09) and automatically reply with FIELD_RESULTS (0x0A). This enables structured request/response workflows alongside normal text commands.

Incoming FIELD_COMMANDS are parsed and routed through the same command registry as text commands, sharing permission checks, type-hinted argument parsing, threading, and middleware.

from lxmfy import LXMFBot, FIELD_COMMANDS, FIELD_RESULTS, pack_result, unpack_commands

bot = LXMFBot(name="FieldBot")

@bot.command(name="status", description="Return bot status")
def status_cmd(ctx):
    # ctx.fields contains the raw LXMF fields dict
    # ctx.request_id is set automatically if the command included one
    ctx.reply("Bot is online")

# Sending a structured command from another LXMF client:
# lxm.fields[FIELD_COMMANDS] = {"command": "status", "args": [], "request_id": "abc123"}
# router.handle_outbound(lxm)

# The bot reply automatically includes FIELD_RESULTS with the response and request_id.

To disable field command processing, set lxmf_commands_enabled=False in BotConfig.

Хранилище¶

The framework provides three storage backends:

JSONStorage¶

from lxmfy import JSONStorage

storage = JSONStorage("data")

SQLiteStorage¶

from lxmfy import SQLiteStorage

storage = SQLiteStorage("data/bot.db")

MemoryStorage¶

from lxmfy.storage import MemoryStorage

storage = MemoryStorage() # Entirely in-memory

Команды¶

Регистрация и обработка команд:

@bot.command(name="hello", description="Says hello")
def hello(ctx):
    ctx.reply(f"Hello {ctx.sender}!")

Type-Hinted Arguments¶

Commands automatically parse and convert arguments based on type hints in the callback function.

@bot.command(name="add", description="Adds two numbers")
def add(ctx, a: int, b: int):
    result = a + b
    ctx.reply(f"The result is {result}")

Help System¶

The framework includes an interactive help generator that provides beautiful, categorized help menus based on Cog and Command metadata.

# The help command is automatically registered.
# Users can use '/help' or '/help <command>'

Многопоточные команды¶

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

import time

@bot.command(name="long_task", description="Performs a long-running task in a separate thread", threaded=True)
def long_task_command(ctx):
    ctx.reply("Starting a long task... please wait.")
    time.sleep(10) # This runs in a separate thread
    ctx.reply("Long task completed!")

Важно: Функции, помеченные как threaded=True, не должны напрямую взаимодействовать с сетевым стеком Reticulum (RNS) или любыми компонентами, которые зависят от lxmfy.transport.py, так как они, как правило, не являются потокобезопасными. Используйте ctx.reply() для отправки сообщений пользователю из многопоточной команды.

События¶

Система событий для обработки различных событий бота:

@bot.events.on("message_received", EventPriority.HIGHEST)
def handle_message(event):
    # Handle message event
    pass

Testing¶

Project tests include reliability and stress scenarios in the repository test suite. Use the repository’s test runner to execute them.

Advanced Reliability Suite¶

The framework includes an extensive suite of automated tests for harsh environments:

Manifold Testing: Validates the mathematical topology of NLP intent vector space.
Chaos Engineering: Simulates bit-rot, SD card failure, and storage corruption.
Temporal Drift: Verifies resilience against system clock jumps (±1 year).
Leak Detection: Long-term tracking of memory, file descriptors, and threads.

Разрешения¶

Система разрешений для контроля доступа к функциям бота:

from lxmfy import DefaultPerms

@bot.command(name="admin", description="Admin command", admin_only=True)
def admin_command(ctx):
    if ctx.is_admin:
        ctx.reply("Admin command executed")

Промежуточное ПО¶

Система промежуточного ПО для обработки сообщений и событий:

@bot.middleware.register(MiddlewareType.PRE_COMMAND)
def pre_command_middleware(ctx):
    # Process before command execution
    pass

Вложения¶

Поддержка отправки файлов, изображений и аудио:

from lxmfy import Attachment, AttachmentType

attachment = Attachment(
    type=AttachmentType.IMAGE,
    name="image.jpg",
    data=image_data,
    format="jpg"
)
bot.send_with_attachment(destination, "Here's an image", attachment)

Внешний вид иконки (поле LXMF)¶

Вы можете установить пользовательскую иконку для своего бота, которую смогут отображать совместимые клиенты LXMF. Для этого используется LXMF.FIELD_ICON_APPEARANCE.

from lxmfy import IconAppearance, pack_icon_appearance_field
import LXMF # Required for LXMF.FIELD_ICON_APPEARANCE

# Define the icon appearance
icon_data = IconAppearance(
    icon_name="smart_toy",  # Name from Material Symbols
    fg_color=b'\xFF\xFF\xFF',  # White foreground (3 bytes)
    bg_color=b'\x4A\x90\xE2'   # Blue background (3 bytes)
)

# Pack it into the LXMF field format
icon_lxmf_field = pack_icon_appearance_field(icon_data)

# Send a message with this icon
bot.send(
    destination_hash_str,
    "Hello from your friendly bot!",
    title="Bot Message",
    lxmf_fields=icon_lxmf_field
)

# You can also combine it with other fields, like attachments:
# attachment_field = pack_attachment(some_attachment)
# combined_fields = {**icon_lxmf_field, **attachment_field}
# bot.send(destination, "Message with icon and attachment", lxmf_fields=combined_fields)

Планировщик¶

Система планирования задач:

@bot.scheduler.schedule(name="daily_task", cron_expr="0 0 * * *")
def daily_task():
    # Run daily at midnight
    pass

Подписи¶

LXMFy предоставляет параметры конфигурации для встроенной в LXMF криптографической подписи и проверки сообщений:

from lxmfy import LXMFBot

bot = LXMFBot(
    name="SecureBot",
    signature_verification_enabled=True,  # Enable signature checks
    require_message_signatures=False      # Set to True to reject unsigned messages
)

Важно: LXMF автоматически обрабатывает все криптографические подписи и проверку с использованием идентификаторов RNS. SignatureManager в LXMFy - это слой конфигурации, который:

Контролирует, следует ли применять проверку подписи
Определяет политику для неподписанных сообщений (принять или отклонить)
Интегрируется с системой разрешений (например, обход проверки для доверенных пользователей)

Фактические криптографические операции выполняются LXMF/RNS, а не LXMFy.

Landlock LSM Sandbox¶

On Linux kernels with Landlock support (5.13+), LXMFy can restrict filesystem access for the bot process and for external script cogs.

Bot process sandbox

When landlock_enabled=True (default) and not running in test_mode, the bot calls apply_landlock_sandbox() during initialization. System directories are read-only; bot storage, config, cogs, Reticulum config, and temp paths remain writable.

bot = LXMFBot(
    name="SecureBot",
    landlock_enabled=True,
)

status = bot.get_landlock_status()
# status keys: landlock_kernel_supported, landlock_requested,
# landlock_auto_enabled, landlock_disabled_by_env, landlock_active

Environment override

LXMFY_LANDLOCK=0: disable Landlock even on supported kernels
LXMFY_LANDLOCK=1: attempt Landlock on Linux regardless of auto-detection
unset: follow landlock_enabled and kernel auto-detection

External cog sandbox

Script cogs use external_cogs_sandbox_type. In auto mode, Landlock is preferred when available because it requires no external tools. See the Creating Bots guide for the full sandbox option list.

Identity Pinning¶

LXMFy supports optional identity pinning to prevent impersonation if an identity is rotated or compromised. When enabled, the bot «pins» an LXMF address to its first-seen public key.

bot = LXMFBot(
    identity_pinning_enabled=True
)

SignatureManager Methods¶

SignatureManager доступен как bot.signature_manager, когда signature_verification_enabled=True:

should_verify_message(sender): Определить, следует ли проверять сообщение от данного отправителя
handle_unsigned_message(sender, message_hash): Handle messages that lack valid signatures based on policy

Как работают подписи LXMF¶

LXMF автоматически подписывает все исходящие сообщения, используя идентификатор RNS отправителя во время операции pack(). При получении сообщений LXMF проверяет подписи и предоставляет:

message.signature_validated: логическое значение, указывающее, действительна ли подпись
message.unverified_reason: код причины, если проверка не удалась (например, SIGNATURE_INVALID, SOURCE_UNKNOWN)

LXMFy использует эти встроенные свойства LXMF для применения политики подписи вашего бота.

Доставка сообщений¶

LXMFy предоставляет расширенные функции доставки сообщений, включая узлы распространения и автоматические повторы:

Узлы распространения¶

Отправляйте сообщения через определенные узлы распространения для повышения надежности в сети Reticulum:

# Configure the propagation node once at config/runtime level
bot.set_propagation_node("<propagation_node_hash>")

# Send using configured delivery behavior
bot.send(
    destination_hash,
    "Message content"
)

# The propagation node hash should be a valid LXMF propagation node
# on the Reticulum network

Автоматические повторы¶

Configure automatic retry attempts for failed direct deliveries:

bot = LXMFBot(
    name="ReliableBot",
    direct_delivery_retries=5,  # Retry direct delivery up to 5 times
    propagation_fallback_enabled=True
)

bot.send(destination_hash, "Important message")

# Default direct_delivery_retries is 3
# Retry logic automatically handles delivery callbacks

Система повторных попыток отслеживает попытки доставки для каждого получателя и автоматически повторяет неудачные доставки. Успешные доставки сбрасывают счетчик повторных попыток для этого получателя.

Message Persistence¶

Outgoing messages can be persisted to disk to ensure they are delivered even after a bot restart. Persistence is enabled by default. The in-memory outbound queue is bounded (message_queue_size, default 50) and drops the oldest message when full. Invalid destination hashes are not restored.

bot = LXMFBot(
    message_persistence_enabled=True,
    message_queue_size=50,
)

Обработчики сообщений¶

LXMFy предоставляет декораторы для обработки различных типов входящих сообщений:

Обработчик первого сообщения¶

Обработка первого сообщения от каждого пользователя:

@bot.on_first_message()
def welcome_user(sender, message):
    content = message.content.decode("utf-8")
    bot.send(sender, f"Welcome! You said: {content}")
    return True  # Return True to stop further processing

Общий обработчик сообщений¶

Обработка всех входящих сообщений перед обработкой команд:

@bot.on_message()
def handle_all_messages(sender, message):
    content = message.content.decode("utf-8").strip()

    # Custom logic here
    if content.startswith("echo:"):
        bot.send(sender, content[5:])
        return True  # Stop further processing

    return False  # Continue to command processing

Обработчики сообщений вызываются в следующем порядке: 1. Обработчик первого сообщения (если это первое сообщение от этого отправителя) 2. Общие обработчики сообщений (зарегистрированные с помощью on_message()) 3. Обработка команд (если сообщение начинается с префикса команды)

Reticulum Relay Chat (RRC)¶

Bots can join RRC hubs over RNS Links with CBOR envelopes. Package: lxmfy.rrc.

BotConfig options¶

rrc_enabled (bool, default False): Connect configured hubs on startup
rrc_hubs (list of hex hashes): Hub destination hashes
rrc_rooms (list of str): Rooms to auto-join after WELCOME
rrc_nick (str or None): Nickname on HELLO and room messages
rrc_dest_name (str, default "rrc.hub"): Destination name used to build the hub destination
rrc_auto_reconnect (bool, default True): Reconnect after link loss
rrc_persist_sessions (bool, default True): Persist hubs and rooms across restarts
reticulum_config_dir (str or None): Reticulum config directory. Also set via LXMFY_RETICULUM_CONFIG_DIR. Use the same config as MeshChatX (often ~/.reticulum) so hub announces are visible.

Example¶

from lxmfy import LXMFBot, RRCMessage

bot = LXMFBot(
    name="RoomBot",
    reticulum_config_dir="~/.reticulum",
    rrc_enabled=True,
    rrc_hubs=["664fc0e8d2e448658e37bb3f34e6c88f"],
    rrc_rooms=["general"],
    rrc_nick="RoomBot",
)

@bot.on_rrc
def on_rrc(event, client, payload):
    if event == "msg" and isinstance(payload, RRCMessage) and payload.mention:
        client.send_message(payload.room, f"Hi {payload.nick}")

# Runtime API
# bot.connect_rrc(hub_hash, rooms=["general"])
# bot.rrc.send_message("general", "hello")
# bot.rrc.send_notice("general", "notice")
# bot.rrc.send_action("general", "waves")
# bot.rrc.join("ops")
# bot.rrc.part("ops")
# bot.rrc.status()
# bot.disconnect_rrc()

Exported types¶

RRCClient: Single-hub session
RRCManager: Multi-hub manager (bot.rrc)
RRCMessage: Room event payload (kind, room, text, nick, src, mention, …)
RRC_VERSION: Wire protocol version constant

Common events passed to @bot.on_rrc handlers include status, welcome, joined, parted, msg, notice, action, motd, error, and rtt.

Шаблоны¶

Фреймворк включает в себя несколько готовых к использованию шаблонов ботов:

EchoBot¶

Простой эхо-бот, который повторяет сообщения:

from lxmfy.templates import EchoBot

bot = EchoBot()
bot.run()

NoteBot¶

Бот для заметок с хранилищем JSON:

from lxmfy.templates import NoteBot

bot = NoteBot()
bot.run()

ReminderBot¶

Бот для напоминаний с хранилищем SQLite:

from lxmfy.templates import ReminderBot

bot = ReminderBot()
bot.run()

RRCBot¶

RRC room bot that joins configured hubs and replies to @mentions. Defaults to hub 664fc0e8d2e448658e37bb3f34e6c88f, room #general, and ~/.reticulum when available.

from lxmfy.templates import RRCBot

bot = RRCBot(
    hubs=["664fc0e8d2e448658e37bb3f34e6c88f"],
    rooms=["general"],
    nick="RRCBot",
    reticulum_config_dir="~/.reticulum",
)
bot.run()

Инструменты командной строки¶

Фреймворк предоставляет инструменты командной строки для управления ботом:

# Create a new bot
lxmfy create mybot

# Create a bot from template
lxmfy create --template echo mybot
lxmfy create --template rrc my_rrc_bot

# Run a template bot
lxmfy run echo
lxmfy run rrc

# Test signature verification with a message
lxmfy signatures test

# Enable signature verification
lxmfy signatures enable

# Disable signature verification
lxmfy signatures disable

Обработка ошибок¶

Фреймворк предоставляет комплексную обработку ошибок:

try:
    bot.run()
except KeyboardInterrupt:
    bot.cleanup()
except Exception as e:
    logger.error(f"Error running bot: {str(e)}")