Adicionar um Adaptador de Plataforma

Adicionar um Adaptador de Plataforma

Este guia cobre como acrescentar uma nova plataforma de mensagens ao gateway do Hermes. Um adaptador de plataforma liga o Hermes a um serviço externo de mensagens (Telegram, Discord, WeCom, etc.) para que as pessoas possam interagir com o agente por aquele serviço.

dica

Há duas maneiras de acrescentar uma plataforma:

• Plugin (recomendado para a comunidade e para terceiros): jogue um diretório de plugin em ~/.hermes/plugins/ — zero mudanças no código do núcleo. Veja Caminho do plugin abaixo.
• Embutida: modifique mais de 20 arquivos entre código, configuração e documentação. Use a Lista de verificação da via embutida abaixo.

Visão geral da arquitetura

Pessoa ↔ Plataforma de mensagens ↔ Adaptador de plataforma ↔ Gateway Runner ↔ AIAgent

Todo adaptador estende BasePlatformAdapter, de gateway/platforms/base.py, e implementa:

• connect() — Estabelece a conexão (WebSocket, long-poll, servidor HTTP, etc.) (abstrato)
• disconnect() — Encerramento limpo (abstrato)
• send() — Envia uma mensagem de texto para uma conversa (abstrato)
• send_typing() — Mostra o indicador de digitação (sobrescrita opcional)
• get_chat_info() — Devolve os metadados da conversa (sobrescrita opcional)

As mensagens de entrada são recebidas pelo adaptador e encaminhadas por self.handle_message(event), que a classe base roteia para o gateway runner.

Caminho do plugin (recomendado)

O sistema de plugins permite acrescentar um adaptador de plataforma sem modificar nenhum código do núcleo do Hermes. Seu plugin é um diretório com dois arquivos:

~/.hermes/plugins/minha-plataforma/
  plugin.yaml      # Metadados do plugin
  adapter.py       # Classe do adaptador + ponto de entrada register()

plugin.yaml

Metadados do plugin. Os blocos requires_env e optional_env preenchem sozinhos as entradas de interface do hermes config (veja Expor variáveis de ambiente abaixo).

name: my-platform
label: My Platform
kind: platform
version: 1.0.0
description: Meu adaptador de plataforma de mensagens personalizada
author: Seu Nome
requires_env:
  - MY_PLATFORM_TOKEN          # string simples funciona
  - name: MY_PLATFORM_CHANNEL  # ou um dicionário rico, para uma experiência melhor
    description: "Canal em que entrar"
    prompt: "Canal"
    password: false
optional_env:
  - name: MY_PLATFORM_HOME_CHANNEL
    description: "Canal padrão para entrega do cron"
    password: false

Ferramentas de cliente de saída: provides_tools

Plugins com kind: platform são adiados: o módulo do adaptador (e os imports do SDK dele) só carregam quando um caminho de gateway, de cron ou de send_message pede a plataforma ao registro de plataformas pela primeira vez. Se o seu plugin também traz ferramentas de cliente de saída que o agente deve poder chamar de qualquer sessão (o a2a_call / a2a_discover etc. do plugin a2a incluído), coloque-as num tools.py dedicado com uma função register_tools(ctx) e declare-as no manifesto:

provides_tools:
  - my_platform_call
  - my_platform_list

Com provides_tools declarado, o Hermes importa apenas o tools.py durante a descoberta de plugins e registra as ferramentas de cliente em todo processo — CLI e TUI inclusive — enquanto o adaptador continua adiado. Mantenha o __init__.py do pacote leve em imports e puxe o adaptador de dentro do register(), para o import antecipado continuar barato. Sem esse campo, nada muda: o plugin inteiro continua adiado.

As pessoas habilitam o conjunto de ferramentas por plataforma como qualquer outro, por exemplo hermes tools enable my_platform --platform cli, ou listando a chave do conjunto sob platform_toolsets no config.yaml. Nomes de plataforma de plugin também são alvos válidos de --platform, então uma sessão de entrada na sua plataforma pode receber as próprias ferramentas de saída.

adapter.py

import os
from gateway.platforms.base import (
    BasePlatformAdapter, SendResult, MessageEvent, MessageType,
)
from gateway.config import Platform, PlatformConfig


class MyPlatformAdapter(BasePlatformAdapter):
    def __init__(self, config: PlatformConfig):
        super().__init__(config, Platform("my_platform"))
        extra = config.extra or {}
        self.token = os.getenv("MY_PLATFORM_TOKEN") or extra.get("token", "")

    async def connect(self, *, is_reconnect: bool = False) -> bool:
        # Conecta na API da plataforma, inicia os ouvintes
        self._mark_connected()
        return True

    async def disconnect(self) -> None:
        self._mark_disconnected()

    async def send(self, chat_id, content, reply_to=None, metadata=None):
        # Envia a mensagem pela API da plataforma
        return SendResult(success=True, message_id="...")

    async def get_chat_info(self, chat_id):
        return {"name": chat_id, "type": "dm"}


def check_requirements() -> bool:
    return bool(os.getenv("MY_PLATFORM_TOKEN"))


def validate_config(config) -> bool:
    extra = getattr(config, "extra", {}) or {}
    return bool(os.getenv("MY_PLATFORM_TOKEN") or extra.get("token"))


def _env_enablement() -> dict | None:
    token = os.getenv("MY_PLATFORM_TOKEN", "").strip()
    channel = os.getenv("MY_PLATFORM_CHANNEL", "").strip()
    if not (token and channel):
        return None
    seed = {"token": token, "channel": channel}
    home = os.getenv("MY_PLATFORM_HOME_CHANNEL")
    if home:
        seed["home_channel"] = {"chat_id": home, "name": "Home"}
    return seed


def register(ctx):
    """Ponto de entrada do plugin — chamado pelo sistema de plugins do Hermes."""
    ctx.register_platform(
        name="my_platform",
        label="My Platform",
        adapter_factory=lambda cfg: MyPlatformAdapter(cfg),
        # Sonda PASSIVA — "as dependências/configurações estão presentes agora?".
        # É chamada por telas de status e pelo carregamento de configuração, então
        # NUNCA pode rodar pip install.
        check_fn=check_requirements,
        # Instalador ATIVO (opcional) — só para plataformas com um SDK que dá para
        # instalar sob demanda. O create_adapter() o chama quando o check_fn
        # devolve False, logo antes de o gateway conectar a plataforma.
        # Costuma envolver tools.lazy_deps.ensure_and_bind(...). Se você omitir,
        # um check_fn False vira um bloqueio definitivo.
        # ensure_deps_fn=ensure_requirements,
        validate_config=validate_config,
        required_env=["MY_PLATFORM_TOKEN"],
        install_hint="pip install my-platform-sdk",
        # Autoconfiguração guiada por ambiente — semeia PlatformConfig.extra a
        # partir de variáveis de ambiente antes de o adaptador ser construído.
        # Veja a seção "Autoconfiguração guiada por variáveis de ambiente" abaixo.
        env_enablement_fn=_env_enablement,
        # Suporte à entrega no canal padrão pelo cron. Faz tarefas de cron com
        # deliver=my_platform serem roteadas sem editar o cron/scheduler.py.
        # Veja a seção "Entrega pelo cron" abaixo.
        cron_deliver_env_var="MY_PLATFORM_HOME_CHANNEL",
        # Variáveis de ambiente de autorização de pessoas, por plataforma
        allowed_users_env="MY_PLATFORM_ALLOWED_USERS",
        allow_all_env="MY_PLATFORM_ALLOW_ALL_USERS",
        # Limite de tamanho da mensagem para o fatiamento inteligente (0 = sem limite)
        max_message_length=4000,
        # Orientação para o LLM, injetada no prompt de sistema
        platform_hint=(
            "You are chatting via My Platform. "
            "It supports markdown formatting."
        ),
        # Exibição
        emoji="💬",
    )

    # Opcional: registrar ferramentas específicas da plataforma
    ctx.register_tool(
        name="my_platform_search",
        toolset="my_platform",
        schema={...},
        handler=my_search_handler,
    )

Configuração

As pessoas configuram a plataforma no config.yaml:

gateway:
  platforms:
    my_platform:
      enabled: true
      extra:
        token: "..."
        channel: "#general"

Ou por variáveis de ambiente (que o adaptador lê no __init__).

O que o sistema de plugins resolve sozinho

Quando você chama ctx.register_platform(), os pontos de integração abaixo são resolvidos para você — sem mudança nenhuma no código do núcleo:

Ponto de integração	Como funciona
Criação do adaptador no gateway	O registro é consultado antes da cadeia if/elif embutida
Leitura da configuração	O Platform._missing_() aceita qualquer nome de plataforma
Validação da plataforma conectada	O validate_config() do registro é chamado
Autorização de pessoas	allowed_users_env / allow_all_env são consultados
Autoativação só por ambiente	O env_enablement_fn semeia PlatformConfig.extra + home_channel
Ponte de configuração YAML	O apply_yaml_config_fn traduz chaves do config.yaml em variáveis de ambiente / extras
Entrega pelo cron	O cron_deliver_env_var faz o deliver=<name> funcionar
Entradas de interface do hermes config	requires_env / optional_env do plugin.yaml preenchem sozinhos
Motor de envio (tools/send_message_tool.py)	Roteia pelo adaptador vivo do gateway
Entrega de webhook entre plataformas	O registro é consultado em busca de plataformas conhecidas
Acesso ao comando /update	Flag allow_update_command
Diretório de canais	Plataformas de plugin entram na enumeração
Dicas no prompt de sistema	O platform_hint é injetado no contexto do LLM
Fatiamento de mensagens	max_message_length para a divisão inteligente
Ocultação de dados pessoais	Flag pii_safe
hermes status	Mostra as plataformas de plugin com a etiqueta (plugin)
hermes gateway setup	As plataformas de plugin aparecem no menu de configuração
hermes tools / hermes skills	Plataformas de plugin na configuração por plataforma
Trava de token (vários perfis)	Use acquire_scoped_lock() no seu connect()
Aviso de configuração órfã	Log descritivo quando o plugin está faltando

Extensões autônomas do caminho de envio

Uma plataforma autônoma pode participar da entrega de saída conduzida pelo host, por hermes send --to ... direto e por deliver=platform:... no cron, declarando o comportamento de envio no mesmo PlatformEntry criado pelo ctx.register_platform(). O send_message deliberadamente não é uma ferramenta de modelo chamável pelo agente; plugins não devem registrar uma superfície de modelo equivalente que deixe o agente iniciar mensagens de saída por conta própria.

async def _send_request(args, chat_id, platform_name, pconfig):
    # `args` contém os campos do pedido de envio conduzido pelo host.
    message_id = await client.send(
        address=chat_id,
        body=args["message"],
        subject=args.get("subject"),
    )
    return {"success": True, "platform": platform_name,
            "chat_id": chat_id, "message_id": message_id}


def _parse_address(raw):
    normalized = raw.strip().lower()
    if normalized.startswith("@") and "@" in normalized[1:]:
        return normalized, None  # (chat_id, thread_id opcional)
    return None                 # segue para a resolução pelo diretório de canais


def _validate_address(address):
    # True aceita; False rejeita; uma string rejeita com aquele diagnóstico.
    return True if address.endswith("@example.com") else "unsupported domain"


def register(ctx):
    ctx.register_platform(
        name="fmsg",
        label="Fixture Message",
        adapter_factory=lambda cfg: FmsgAdapter(cfg),
        check_fn=check_requirements,
        parse_target_ref_fn=_parse_address,
        validate_target_ref_fn=_validate_address,
        # Pode ser uma função comum ou async def. O Hermes aguarda qualquer
        # resultado aguardável, incluindo objetos chamáveis e invólucros
        # functools.partial.
        send_message_handler=_send_request,
        # Prefira este gancho de nível mais baixo quando o cron precisar enviar
        # de um processo sem o gateway vivo.
        standalone_sender_fn=_standalone_send,
    )

A resolução de alvo é compartilhada pelas três superfícies de saída. A saída do analisador é normalizada primeiro, e os IDs do diretório de canais são considerados confiáveis. Um analisador de plugin precisa aceitar explicitamente a sintaxe nativa de alvo; strings não resolvidas nunca passam adiante de forma opaca. Plataformas desconhecidas e falhas do validador devolvem um diagnóstico em vez de tentar a entrega em silêncio. Recarregamentos forçados de plugin e trocas de perfil cancelam o registro das entradas que pertencem a eles, então analisadores e tratadores não vazam para o próximo perfil.

Autoconfiguração guiada por variáveis de ambiente

A maioria das pessoas configura uma plataforma jogando variáveis de ambiente no ~/.hermes/.env, em vez de editar o config.yaml. O gancho env_enablement_fn deixa seu plugin capturar essas variáveis antes de o adaptador ser construído, então hermes gateway status, get_connected_platforms() e a entrega pelo cron enxergam o estado correto sem instanciar o SDK da plataforma.

def _env_enablement() -> dict | None:
    """Semeia PlatformConfig.extra a partir de variáveis de ambiente.

    Chamada pelo registro de plataformas durante load_gateway_config().
    Devolve None quando a plataforma não está minimamente configurada — quem
    chamou então pula a autoativação. Devolva um dicionário para semear os extras.

    A chave especial 'home_channel' é extraída e vira uma dataclass HomeChannel
    de verdade no PlatformConfig; toda outra chave é mesclada em
    PlatformConfig.extra.
    """
    token = os.getenv("MY_PLATFORM_TOKEN", "").strip()
    channel = os.getenv("MY_PLATFORM_CHANNEL", "").strip()
    if not (token and channel):
        return None
    seed = {"token": token, "channel": channel}
    home = os.getenv("MY_PLATFORM_HOME_CHANNEL")
    if home:
        seed["home_channel"] = {
            "chat_id": home,
            "name": os.getenv("MY_PLATFORM_HOME_CHANNEL_NAME", "Home"),
        }
    return seed


def register(ctx):
    ctx.register_platform(
        name="my_platform",
        label="My Platform",
        adapter_factory=lambda cfg: MyPlatformAdapter(cfg),
        check_fn=check_requirements,
        validate_config=validate_config,
        env_enablement_fn=_env_enablement,
        # ... outros campos
    )

Ponte de configuração YAML→ambiente

Algumas pessoas preferem definir chaves do config.yaml (my_platform.require_mention, my_platform.allowed_channels, etc.) a usar variáveis de ambiente. O gancho apply_yaml_config_fn deixa seu plugin ser dono dessa tradução, em vez de obrigar o gateway/config.py do núcleo a conhecer o esquema YAML da sua plataforma.

import os

def _apply_yaml_config(yaml_cfg: dict, platform_cfg: dict) -> dict | None:
    """Traduz as chaves `my_platform:` do config.yaml em variáveis de ambiente / extras.

    yaml_cfg     — o dicionário completo do config.yaml lido no nível de cima
    platform_cfg — o subdicionário da própria plataforma (yaml_cfg.get("my_platform", {}))

    Pode alterar o os.environ diretamente (use travas com `not os.getenv(...)`
    para preservar a precedência ambiente > YAML) e/ou devolver um dicionário
    para mesclar em PlatformConfig.extra. Devolva None ou {} se não houver extras.
    """
    if "require_mention" in platform_cfg and not os.getenv("MY_PLATFORM_REQUIRE_MENTION"):
        os.environ["MY_PLATFORM_REQUIRE_MENTION"] = str(platform_cfg["require_mention"]).lower()
    allowed = platform_cfg.get("allowed_channels")
    if allowed is not None and not os.getenv("MY_PLATFORM_ALLOWED_CHANNELS"):
        if isinstance(allowed, list):
            allowed = ",".join(str(v) for v in allowed)
        os.environ["MY_PLATFORM_ALLOWED_CHANNELS"] = str(allowed)
    return None  # nada de extra para mesclar em PlatformConfig.extra

def register(ctx):
    ctx.register_platform(
        name="my_platform",
        ...,
        apply_yaml_config_fn=_apply_yaml_config,
    )

O gancho é invocado durante o load_gateway_config(), depois do laço genérico de chaves compartilhadas (que cuida de chaves comuns como unauthorized_dm_behavior, notice_delivery, reply_prefix, require_mention, etc.) e antes do _apply_env_overrides(), então seu plugin só precisa fazer a ponte das chaves específicas da plataforma.

Exceções levantadas pelo gancho são engolidas e registradas em nível de depuração — um plugin mal comportado nunca aborta o carregamento da configuração do gateway.

Entrega pelo cron

Para que tarefas de cron com deliver=my_platform sejam roteadas para um canal padrão configurado, defina cron_deliver_env_var com o nome da variável de ambiente que guarda o ID padrão de conversa/sala/canal:

ctx.register_platform(
    name="my_platform",
    ...
    cron_deliver_env_var="MY_PLATFORM_HOME_CHANNEL",
)

O agendador lê essa variável de ambiente ao resolver o alvo padrão das tarefas com deliver=my_platform, e também trata a plataforma como um alvo válido de cron nas verificações do tipo _KNOWN_DELIVERY_PLATFORMS. Se o seu env_enablement_fn semear um dicionário home_channel (veja acima), ele tem precedência — o cron_deliver_env_var é a reserva para tarefas de cron que rodam antes da semeadura por ambiente.

Entrega pelo cron fora do processo

O cron_deliver_env_var faz sua plataforma virar um alvo reconhecido de deliver=. Para que o envio de fato dê certo quando a tarefa de cron roda num processo separado do gateway (ou seja, hermes cron run separado do hermes gateway), registre um standalone_sender_fn:

async def _standalone_send(
    pconfig,
    chat_id,
    message,
    *,
    thread_id=None,
    media_files=None,
    force_document=False,
):
    """Abre uma conexão efêmera / obtém um token novo, envia e fecha."""
    # ... abre a conexão, envia a mensagem, devolve o resultado ...
    return {"success": True, "message_id": "..."}
    # ou {"error": "..."}

ctx.register_platform(
    name="my_platform",
    ...
    cron_deliver_env_var="MY_PLATFORM_HOME_CHANNEL",
    standalone_sender_fn=_standalone_send,
)

Por que esse gancho é necessário: as plataformas embutidas (Telegram, Discord, Slack, etc.) trazem auxiliares REST diretos em tools/send_message_tool.py, para o cron conseguir entregar sem manter o gateway no mesmo processo. As plataformas de plugin historicamente dependiam do _gateway_runner_ref(), que devolve None fora do processo do gateway; então, sem o standalone_sender_fn, o envio pelo lado do cron falha com No live adapter for platform '<name>'.

A função recebe o mesmo pconfig e o mesmo chat_id que o adaptador vivo receberia, além dos argumentos nomeados opcionais thread_id, media_files e force_document. Devolver {"success": True, "message_id": ...} é tratado como entrega bem-sucedida; devolver {"error": "..."} faz a mensagem aparecer no delivery_errors do cron. Exceções levantadas dentro da função são capturadas pelo despachante e relatadas como Plugin standalone send failed: <reason>. Implementações de referência ficam em plugins/platforms/{irc,teams,google_chat}/adapter.py.

Expor variáveis de ambiente no hermes config

O hermes_cli/config.py varre plugins/platforms/*/plugin.yaml no momento do import e preenche sozinho o OPTIONAL_ENV_VARS a partir dos blocos requires_env e (opcional) optional_env. Use a forma de dicionário rico para contribuir com descrições, prompts, marcações de senha e URLs adequados — a interface de configuração da CLI aproveita tudo de graça.

# plugins/platforms/my_platform/plugin.yaml
name: my_platform-platform
label: My Platform
kind: platform
version: 1.0.0
description: >
  Adaptador de gateway da My Platform para o Hermes Agent.
author: Seu Nome
requires_env:
  - name: MY_PLATFORM_TOKEN
    description: "Token da API de bot, obtido no console da My Platform"
    prompt: "Token de bot da My Platform"
    url: "https://my-platform.example.com/bots"
    password: true
  - name: MY_PLATFORM_CHANNEL
    description: "Canal em que entrar (por exemplo, #hermes)"
    prompt: "Canal"
    password: false
optional_env:
  - name: MY_PLATFORM_HOME_CHANNEL
    description: "Canal padrão para entrega do cron (por padrão, MY_PLATFORM_CHANNEL)"
    prompt: "Canal padrão (ou vazio)"
    password: false
  - name: MY_PLATFORM_ALLOWED_USERS
    description: "IDs de pessoas autorizadas a falar com o bot, separados por vírgula"
    prompt: "Pessoas autorizadas (separadas por vírgula)"
    password: false

Chaves de dicionário aceitas: name (obrigatória), description, prompt, url, password (booleano; detectado sozinho pelo sufixo *_TOKEN / *_SECRET / *_KEY / *_PASSWORD / *_JSON quando omitido), category (por padrão, "messaging").

Entradas como string simples (- MY_PLATFORM_TOKEN) continuam funcionando — elas ganham uma descrição genérica derivada do label do plugin. Se já existe uma entrada fixa no código para a mesma variável em OPTIONAL_ENV_VARS, ela vence (compatibilidade com o que já existia); a forma do plugin.yaml funciona como reserva.

Experiência de LLM lento específica de plataforma

Algumas plataformas têm restrições que mudam como uma resposta lenta do LLM deve ser apresentada:

• O LINE emite um reply token de uso único que expira mais ou menos 60 segundos depois do evento de entrada. Responder com esse token é gratuito; recorrer à Push API tarifada não é. Se o LLM não terminou até o prazo, a escolha é "queimar cota paga de Push" ou "fazer algo mais esperto com o reply token antes de ele expirar".
• O WhatsApp marca uma sessão como inativa depois de 24h, e a partir daí só aceita mensagens de template.
• O SMS não tem noção de indicador de digitação nem de atualizações progressivas — respostas longas simplesmente parecem um bot fora do ar.

Essas são restrições reais que o BasePlatformAdapter base não tem como prever. A superfície de plugin deixa espaço de propósito para um adaptador acrescentar uma camada de experiência específica da plataforma em cima do laço de digitação base, sem inchar a lista de argumentos nomeados.

Padrão: sobrescrever _keep_typing para acrescentar experiência durante a execução

O BasePlatformAdapter._keep_typing é o batimento do indicador de digitação — ele roda como tarefa de segundo plano enquanto o LLM gera, e é cancelado quando a resposta é entregue. Para acrescentar um comportamento específico da plataforma num limiar (por exemplo, enviar um balão de "ainda pensando" aos 45s), sobrescreva o _keep_typing no seu adaptador, agende a sua própria tarefa ao lado do super()._keep_typing() e derrube tudo no finally:

class LineAdapter(BasePlatformAdapter):
    async def _keep_typing(self, chat_id: str, *args, **kwargs) -> None:
        if self.slow_response_threshold <= 0:
            await super()._keep_typing(chat_id, *args, **kwargs)
            return

        async def _fire_at_threshold() -> None:
            try:
                await asyncio.sleep(self.slow_response_threshold)
            except asyncio.CancelledError:
                raise
            # Trabalho específico da plataforma aqui — no LINE, envie um balão
            # de Template Buttons "Get answer" usando o reply token em cache,
            # para a pessoa poder buscar a resposta guardada depois, com um
            # reply token novo (e gratuito) vindo do callback de postback.
            await self._send_slow_response_button(chat_id)

        side_task = asyncio.create_task(_fire_at_threshold())
        try:
            await super()._keep_typing(chat_id, *args, **kwargs)
        finally:
            if not side_task.done():
                side_task.cancel()
                try:
                    await side_task
                except (asyncio.CancelledError, Exception):
                    pass

Pontos-chave:

• Sempre faça await super()._keep_typing(...). O batimento de digitação é útil por conta própria — não substitua, acrescente por cima.
• Derrube a tarefa paralela no finally. Quando o LLM termina (ou o /stop cancela a execução), o gateway cancela a tarefa de digitação. Sua tarefa paralela precisa observar esse cancelamento também, senão ela fica pendurada e pode disparar depois de a resposta já ter sido entregue.
• Combine com o interrupt_session_activity para resolver qualquer estado órfão de interface quando a pessoa manda /stop. No LINE, isso significa passar a entrada do cache de postback de PENDING para ERROR, para o botão permanente "Get answer" entregar uma mensagem de "a execução foi interrompida" em vez de ficar em laço.

Padrão: sobrescrever o send para passar por um cache em vez de enviar na hora

Se a sua experiência de resposta lenta guarda a resposta em cache para busca posterior (o fluxo de postback do LINE), a sua sobrescrita de send precisa reconhecer três modos:

1. Postback pendente ativo para esta conversa → guarde a resposta sob o request_id e não envie nada visível.
2. Confirmação de ocupado do sistema (⚡ Interrupting, ⏳ Queued, ⏩ Steered) → desvie do cache e envie de forma visível, para a pessoa ver a resposta do gateway à entrada dela.
3. Resposta normal → envie por reply token ou push, como de costume.

async def send(self, chat_id: str, content: str, **kw) -> SendResult:
    if _is_system_bypass(content):
        return await self._send_text_chunks(chat_id, content, force_push=False)
    pending_rid = self._pending_buttons.get(chat_id)
    if pending_rid:
        self._cache.set_ready(pending_rid, content)
        return SendResult(success=True, message_id=pending_rid)
    return await self._send_text_chunks(chat_id, content, force_push=False)

Os _SYSTEM_BYPASS_PREFIXES são os prefixos de confirmação de ocupado do próprio gateway (⚡, ⏳, ⏩, 💾). Deixe sempre esses passarem de forma visível, independentemente do estado da interface em cache.

Quando esse padrão faz sentido

Use a abordagem de sobrescrever o laço de digitação quando:

• A API de saída da plataforma tem uma restrição rígida de janela de tempo (reply token de uso único, sessão fixa que expira, etc.) E
• Um balão visível no meio da execução é uma experiência aceitável naquela plataforma.

Use o caminho mais simples de slow_response_threshold = 0, sempre com Push, quando:

• A plataforma não tem uma distinção relevante entre gratuito e pago, OU
• A comunidade prefere o silêncio de "carregando… carregando… PRONTO" seguido da resposta a um balão intermediário interativo.

O LINE aceita os dois: o limiar é 45s por padrão, para a busca gratuita por postback, e LINE_SLOW_RESPONSE_THRESHOLD=0 volta ao "sempre recorrer ao Push".

Implementação de referência

Veja plugins/platforms/line/adapter.py para a implementação completa de postback do LINE — uma máquina de estados RequestCache (PENDING → READY → DELIVERED, mais ERROR para o /stop), uma sobrescrita de _keep_typing que dispara o balão de Template Buttons no limiar, uma sobrescrita de send que passa pelo cache, e uma sobrescrita de interrupt_session_activity que resolve entradas PENDING órfãs.

Implementações de referência (caminho do plugin)

Veja plugins/platforms/irc/ no repositório para um exemplo completo e funcional — um adaptador de IRC totalmente assíncrono, com zero dependências externas. O plugins/platforms/teams/ cobre Bot Framework / Adaptive Cards, o plugins/platforms/google_chat/ cobre APIs REST baseadas em OAuth, e o plugins/platforms/line/ cobre APIs de mensagens guiadas por webhook com experiência de LLM lento específica da plataforma.

---

Lista de verificação passo a passo (via embutida)

observação

Esta lista é para acrescentar uma plataforma direto no código do núcleo do Hermes — normalmente feito por quem contribui com o núcleo, para plataformas com suporte oficial. Plataformas da comunidade ou de terceiros devem usar o Caminho do plugin acima.

1. Enum de plataforma

Acrescente sua plataforma ao enum Platform em gateway/config.py:

class Platform(Enum):
    # ... plataformas existentes ...
    NEWPLAT = "newplat"

2. Arquivo do adaptador

Crie plugins/platforms/newplat/adapter.py:

from gateway.config import Platform, PlatformConfig
from gateway.platforms.base import (
    BasePlatformAdapter, MessageEvent, MessageType, SendResult,
)

def check_newplat_requirements() -> bool:
    """Devolve True se as dependências estiverem disponíveis."""
    return SOME_SDK_AVAILABLE

class NewPlatAdapter(BasePlatformAdapter):
    def __init__(self, config: PlatformConfig):
        super().__init__(config, Platform.NEWPLAT)
        # Lê a configuração do dicionário config.extra
        extra = config.extra or {}
        self._api_key = extra.get("api_key") or os.getenv("NEWPLAT_API_KEY", "")

    async def connect(self, *, is_reconnect: bool = False) -> bool:
        # Prepara a conexão, inicia o polling/webhook
        self._mark_connected()
        return True

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

    async def send(self, chat_id, content, reply_to=None, metadata=None):
        # Envia a mensagem pela API da plataforma
        return SendResult(success=True, message_id="...")

    async def get_chat_info(self, chat_id):
        return {"name": chat_id, "type": "dm"}

Para mensagens de entrada, monte um MessageEvent e chame self.handle_message(event):

source = self.build_source(
    chat_id=chat_id,
    chat_name=name,
    chat_type="dm",  # ou "group"
    user_id=user_id,
    user_name=user_name,
)
event = MessageEvent(
    text=content,
    message_type=MessageType.TEXT,
    source=source,
    message_id=msg_id,
)
await self.handle_message(event)

3. Configuração do gateway (gateway/config.py)

Três pontos de contato:

1. get_connected_platforms() — Acrescente uma verificação das credenciais exigidas pela sua plataforma
2. load_gateway_config() — Acrescente a entrada no mapa de tokens de ambiente: Platform.NEWPLAT: "NEWPLAT_TOKEN"
3. _apply_env_overrides() — Mapeie todas as variáveis de ambiente NEWPLAT_* para a configuração

4. Gateway Runner (gateway/run.py)

Seis pontos de contato:

1. _create_adapter() — Acrescente um ramo elif platform == Platform.NEWPLAT:
2. Mapa allowed_users do _is_user_authorized() — Platform.NEWPLAT: "NEWPLAT_ALLOWED_USERS"
3. Mapa allow_all do _is_user_authorized() — Platform.NEWPLAT: "NEWPLAT_ALLOW_ALL_USERS"
4. Tupla _any_allowlist da verificação antecipada de ambiente — Acrescente "NEWPLAT_ALLOWED_USERS"
5. Tupla _allow_all da verificação antecipada de ambiente — Acrescente "NEWPLAT_ALLOW_ALL_USERS"
6. frozenset _UPDATE_ALLOWED_PLATFORMS — Acrescente Platform.NEWPLAT

5. Entrega entre plataformas

1. gateway/platforms/webhook.py — Acrescente "newplat" à tupla de tipos de entrega
2. cron/scheduler.py — Acrescente ao frozenset _KNOWN_DELIVERY_PLATFORMS e ao mapa de plataformas do _deliver_result()

6. Integração com a CLI

1. hermes_cli/config.py — Acrescente todas as variáveis NEWPLAT_* ao _EXTRA_ENV_KEYS
2. hermes_cli/gateway.py — Acrescente a entrada à lista _PLATFORMS, com key, label, emoji, token_var, setup_instructions e vars
3. hermes_cli/platforms.py — Acrescente a entrada PlatformInfo com label e default_toolset (usados pelas TUIs de skills_config e tools_config)
4. hermes_cli/setup.py — Acrescente a função _setup_newplat() (pode delegar para o gateway.py) e acrescente a tupla à lista de plataformas de mensagens
5. hermes_cli/status.py — Acrescente a entrada de detecção de plataforma: "NewPlat": ("NEWPLAT_TOKEN", "NEWPLAT_HOME_CHANNEL")
6. hermes_cli/dump.py — Acrescente "newplat": "NEWPLAT_TOKEN" ao dicionário de detecção de plataformas

7. Ferramentas

1. tools/send_message_tool.py — Acrescente "newplat": Platform.NEWPLAT ao mapa de plataformas
2. tools/cronjob_tools.py — Acrescente newplat à string de descrição dos alvos de entrega

8. Conjuntos de ferramentas

1. toolsets.py — Acrescente a definição do conjunto "hermes-newplat" com _HERMES_CORE_TOOLS
2. toolsets.py — Acrescente "hermes-newplat" à lista de includes de "hermes-gateway"

9. Opcional: dicas de plataforma

agent/prompt_builder.py — Se a sua plataforma tem limitações específicas de renderização (sem markdown, limite de tamanho de mensagem, etc.), acrescente uma entrada ao dicionário PLATFORM_HINTS. Isso injeta orientação específica da plataforma no prompt de sistema:

PLATFORM_HINTS = {
    # ...
    "newplat": (
        "You are chatting via NewPlat. It supports markdown formatting "
        "but has a 4000-character message limit."
    ),
}

Nem toda plataforma precisa de dicas — só acrescente uma se o comportamento do agente deve mudar.

10. Testes

Crie tests/gateway/test_newplat.py cobrindo:

• Construção do adaptador a partir da configuração
• Montagem do evento de mensagem
• Método de envio (com a API externa simulada)
• Recursos específicos da plataforma (criptografia, roteamento, etc.)

11. Documentação

Arquivo	O que acrescentar
website/docs/user-guide/messaging/newplat.md	Página completa de configuração da plataforma
website/docs/user-guide/messaging/index.md	Tabela de comparação de plataformas, diagrama de arquitetura, tabela de conjuntos de ferramentas, seção de segurança, link de próximos passos
website/docs/reference/environment-variables.md	Todas as variáveis de ambiente NEWPLAT_*
website/docs/reference/toolsets-reference.md	Conjunto de ferramentas hermes-newplat
website/docs/integrations/index.md	Link da plataforma
website/sidebars.ts	Entrada de barra lateral para a página de documentação
website/docs/developer-guide/architecture.md	Contagem e listagem de adaptadores
website/docs/developer-guide/gateway-internals.md	Listagem do arquivo de adaptador

Auditoria de paridade

Antes de dar um PR de nova plataforma por concluído, rode uma auditoria de paridade contra uma plataforma consolidada:

# Encontre todo arquivo .py que menciona a plataforma de referência
search_files "bluebubbles" output_mode="files_only" file_glob="*.py"

# Encontre todo arquivo .py que menciona a nova plataforma
search_files "newplat" output_mode="files_only" file_glob="*.py"

# Todo arquivo do primeiro conjunto que não está no segundo é uma lacuna em potencial

Repita para os arquivos .md e .ts. Investigue cada lacuna — é uma enumeração de plataformas (precisa ser atualizada) ou uma referência específica de plataforma (dá para pular)?

Padrões comuns

Adaptadores de long-poll

Se o seu adaptador usa long-polling (como o Telegram ou o Weixin), use uma tarefa com laço de sondagem:

async def connect(self):
    self._poll_task = asyncio.create_task(self._poll_loop())
    self._mark_connected()

async def _poll_loop(self):
    while self._running:
        messages = await self._fetch_updates()
        for msg in messages:
            await self.handle_message(self._build_event(msg))

Adaptadores de callback/webhook

Se a plataforma empurra as mensagens para o seu endpoint (como o WeCom Callback), rode um servidor HTTP:

async def connect(self):
    self._app = web.Application()
    self._app.router.add_post("/callback", self._handle_callback)
    # ... inicia o servidor aiohttp
    self._mark_connected()

async def _handle_callback(self, request):
    event = self._build_event(await request.text())
    await self._message_queue.put(event)
    return web.Response(text="success")  # Confirma imediatamente

Para plataformas com prazos apertados de resposta (por exemplo, o limite de 5 segundos do WeCom), confirme sempre na hora e entregue a resposta do agente por iniciativa própria, pela API, mais tarde. Sessões de agente rodam de 3 a 30 minutos — respostas embutidas dentro da janela de resposta de um callback não são viáveis.

Travas de token

Se o adaptador mantém uma conexão persistente com uma credencial única, acrescente uma trava com escopo para impedir que dois perfis usem a mesma credencial:

from gateway.status import acquire_scoped_lock, release_scoped_lock

async def connect(self, *, is_reconnect: bool = False):
    acquired, _existing = acquire_scoped_lock("newplat", self._token)
    if not acquired:
        logger.error("Token already in use by another profile")
        return False
    # ... conecta

async def disconnect(self):
    release_scoped_lock("newplat", self._token)

Implementações de referência

Adaptador	Padrão	Complexidade	Boa referência para
bluebubbles.py	REST + webhook	Média	Integração simples com API REST
weixin.py	Long-poll + CDN	Alta	Tratamento de mídia, criptografia
plugins/platforms/wecom/callback_adapter.py	Callback/webhook	Média	Servidor HTTP, cripto AES, múltiplos apps
plugins/platforms/irc/adapter.py	Long-poll + protocolo IRC	Alta	Adaptador de plugin completo com trava de token com escopo

---