11 KiB
Surface Protocol — унификация поверхностей
Идея
Любая поверхность (Telegram, Matrix, будущий Discord, голос, web) делает одно и то же: принимает события от пользователя и отправляет ответы обратно.
Разница только в том как событие выглядит снаружи: aiogram Message,
matrix-nio RoomMessageText, Discord Interaction. Внутри — одно и то же.
Surface Protocol — это общий язык между адаптерами поверхностей и ядром. Адаптер конвертирует нативные события в протокольные структуры и обратно. Ядро работает только с протокольными структурами и ничего не знает о транспорте.
Добавить новую поверхность = написать один адаптер-конвертер. Ядро не трогается.
Структура проекта
surfaces-bot/
core/
protocol.py — все унифицированные структуры данных
handler.py — EventDispatcher: IncomingEvent → OutgoingEvent
handlers/ — обработчики по типам событий (start, message, chat, settings, callback)
store.py — StateStore Protocol + InMemoryStore + SQLiteStore
chat.py — ChatManager: метаданные чатов C1/C2/C3
auth.py — AuthManager, AuthFlow
settings.py — SettingsManager, SettingsAction
adapter/
telegram/ — aiogram адаптер
converter.py — aiogram Event → IncomingEvent, OutgoingEvent → aiogram API
bot.py — точка входа, роутер
matrix/ — matrix-nio адаптер
converter.py — matrix-nio Event → IncomingEvent, OutgoingEvent → Matrix API
bot.py — точка входа, клиент
sdk/
interface.py — Protocol: PlatformClient (контракт к SDK)
real.py — RealPlatformClient (через AgentApi)
mock.py — MockPlatformClient (для локальных тестов)
Входящие события (Incoming)
Всё что пользователь делает конвертируется в одно из трёх:
IncomingMessage
Обычное сообщение — текст, файл, голос.
@dataclass
class IncomingMessage:
user_id: str # "tg_123456" | "@user:matrix.org"
platform: str # "telegram" | "matrix"
chat_id: str # "C1" | "C2" — ID чата в воркспейсе платформы
text: str
attachments: list[Attachment]
reply_to: str | None # message_id если это ответ на сообщение
IncomingCommand
Команда управления — /start, !new, /settings.
@dataclass
class IncomingCommand:
user_id: str
platform: str
chat_id: str
command: str # "start" | "new" | "settings" | "archive" ...
args: list[str] # аргументы после команды
IncomingCallback
Нажатие кнопки или реакции — подтверждение, переключение скилла.
@dataclass
class IncomingCallback:
user_id: str
platform: str
chat_id: str
action: str # "confirm" | "cancel" | "toggle_skill" | "connect" ...
payload: dict # {"skill": "browser"} | {"action_id": "abc123"} ...
Attachment
Вложение, нормализованное из любого источника.
@dataclass
class Attachment:
type: str # "image" | "document" | "audio" | "video"
url: str | None # ссылка если доступна
content: bytes | None # содержимое если скачано
filename: str | None
mime_type: str | None
Исходящие события (Outgoing)
Всё что ядро хочет показать пользователю:
OutgoingMessage
Ответ агента или системное сообщение.
@dataclass
class OutgoingMessage:
chat_id: str
text: str
parse_mode: str # "markdown" | "plain"
attachments: list[Attachment]
reply_to: str | None
OutgoingUI
Интерактивные элементы — кнопки, меню, переключатели.
@dataclass
class OutgoingUI:
chat_id: str
text: str
buttons: list[UIButton]
@dataclass
class UIButton:
label: str
action: str # action для IncomingCallback
payload: dict
style: str # "primary" | "danger" | "secondary"
Telegram рендерит это как InlineKeyboard. Matrix рендерит как текст (в MVP).
OutgoingNotification
Асинхронное уведомление — агент закончил долгую задачу.
@dataclass
class OutgoingNotification:
chat_id: str
text: str
level: str # "info" | "warning" | "success" | "error"
OutgoingTyping
Индикатор что агент думает.
@dataclass
class OutgoingTyping:
chat_id: str
is_typing: bool
Жизненный цикл (Lifecycle)
Унифицированные события для управления чатами и подключением.
ChatContext
Метаданные чата — общие для всех поверхностей. Хранятся ботом, lifecycle контейнера управляет платформа (Master).
@dataclass
class ChatContext:
chat_id: str # "C1" | "C2" — ID чата в воркспейсе платформы
display_name: str # «Чат 1» | «Анализ рынка»
platform: str
surface_ref: str # room_id в Matrix | topic_id в Telegram
created_at: datetime
is_archived: bool
AuthFlow
Флоу аутентификации — одинаков для всех поверхностей.
@dataclass
class AuthFlow:
user_id: str
platform: str
state: str # "pending" | "code_sent" | "confirmed" | "failed"
platform_user_id: str | None
ConfirmationRequest
Запрос подтверждения опасного действия — от агента к пользователю.
@dataclass
class ConfirmationRequest:
action_id: str
chat_id: str
description: str # «Отправить письмо на vasya@mail.ru»
risk_level: str # "low" | "medium" | "high"
expires_at: datetime
Telegram показывает как Inline-кнопки.
Matrix показывает как запрос для !yes / !no.
Ядро не знает как именно — только получает IncomingCallback с action: "confirm".
SettingsAction
Унифицированные действия в настройках — одинаковы для Telegram и Matrix.
@dataclass
class SettingsAction:
action: str # "connect" | "disconnect" | "toggle_skill" | ...
payload: dict
# Примеры:
SettingsAction(action="connect", payload={"service": "gmail"})
SettingsAction(action="toggle_skill", payload={"skill": "browser", "enabled": True})
SettingsAction(action="set_soul", payload={"field": "name", "value": "Лямбда"})
SettingsAction(action="set_safety", payload={"trigger": "email-send", "enabled": True})
Ядро обрабатывает SettingsAction одинаково. Откуда пришло — не важно.
PaymentEvent
Заглушка для биллинга — реализует другая команда.
@dataclass
class PaymentRequired:
user_id: str
reason: str # "limit_reached" | "feature_locked"
current_plan: str
Поверхность получила PaymentRequired → показала заглушку.
Содержимое заглушки потом наполняет команда биллинга.
Как написать новую поверхность
Скопируй adapter/_template.py и реализуй три метода:
class MySurfaceAdapter:
def parse_incoming(self, raw_event) -> IncomingMessage | IncomingCommand | IncomingCallback:
"""Конвертировать нативное событие в протокольную структуру."""
...
def render_outgoing(self, event: OutgoingMessage | OutgoingUI | OutgoingNotification) -> any:
"""Конвертировать протокольную структуру в нативный формат."""
...
async def send(self, rendered) -> None:
"""Отправить нативным способом."""
...
Дальше подключи адаптер в точку входа и передай core.handler.handle как callback.
Всё остальное уже работает.
Что ядро даёт бесплатно
Любая новая поверхность получает без дополнительного кода:
- управление чатами (
ChatContext, C1/C2/C3) - аутентификацию (
AuthFlow) - подтверждение действий (
ConfirmationRequest) - все настройки (коннекторы, скиллы, SOUL, безопасность, подписка)
- интеграцию с платформой через
PlatformClient - обработку ошибок платформы
Замена MockPlatformClient на реальный SDK
Вся работа с платформой идёт через PlatformClient протокол:
class PlatformClient(Protocol):
async def get_or_create_user(self, external_id: str, platform: str,
display_name: str | None = None) -> User: ...
async def send_message(self, user_id: str, chat_id: str, text: str,
attachments: list | None = None) -> MessageResponse: ...
async def get_settings(self, user_id: str) -> UserSettings: ...
async def update_settings(self, user_id: str, action: Any) -> None: ...
Бот не управляет lifecycle контейнеров агентов. Запуск/перезапуск агентов — ответственность платформы.
Бот передаёт user_id + chat_id + текст.
MockPlatformClient реализует этот протокол для локальных тестов.
Реальный SDK используется через RealPlatformClient (sdk/real.py), который подключается к AgentApi по WebSocket.
Адаптеры поверхностей и ядро не меняются вообще, привязка идёт через config/matrix-agents.yaml.