Adding a Platform Adapter | Hermes Agent
Skip to main content

Adding a Platform Adapter

This guide covers adding a new messaging platform to the Hermes gateway. A platform adapter connects Hermes to an external messaging service (Telegram, Discord, WeCom, etc.) so users can interact with the agent through that service.

tip

There are two ways to add a platform:

  • Plugin (recommended for community/third-party): Drop a plugin directory into ~/.hermes/plugins/ — zero core code changes needed. See Plugin Path below.
  • Built-in: Modify 20+ files across code, config, and docs. Use the Built-in Checklist below.

Architecture Overview​

User ↔ Messaging Platform ↔ Platform Adapter ↔ Gateway Runner ↔ AIAgent

Every adapter extends BasePlatformAdapter from gateway/platforms/base.py and implements:

  • connect() — Establish connection (WebSocket, long-poll, HTTP server, etc.) (abstract)
  • disconnect() — Clean shutdown (abstract)
  • send() — Send a text message to a chat (abstract)
  • send_typing() — Show typing indicator (optional override)
  • get_chat_info() — Return chat metadata (optional override)

Inbound messages are received by the adapter and forwarded via self.handle_message(event), which the base class routes to the gateway runner.

The plugin system lets you add a platform adapter without modifying any core Hermes code. Your plugin is a directory with two files:

~/.hermes/plugins/my-platform/
  plugin.yaml      # Plugin metadata
  adapter.py       # Adapter class + register() entry point

plugin.yaml​

Plugin metadata. The requires_env and optional_env blocks auto-populate hermes config UI entries (see Surfacing Env Vars below).

name: my-platform
label: My Platform
kind: platform
version: 1.0.0
description: My custom messaging platform adapter
author: Your Name
requires_env:
  - MY_PLATFORM_TOKEN          # bare string works
  - name: MY_PLATFORM_CHANNEL  # or rich dict for better UX
    description: "Channel to join"
    prompt: "Channel"
    password: false
optional_env:
  - name: MY_PLATFORM_HOME_CHANNEL
    description: "Default channel for cron delivery"
    password: false

Outbound client tools: provides_tools​

kind: platform plugins are deferred: the adapter module (and its SDK imports) only load when a gateway, cron, or send_message path first asks the platform registry for the platform. If your plugin also ships outbound client tools the agent should be able to call from any session (the bundled a2a plugin's a2a_call / a2a_discover etc.), put them in a dedicated tools.py with a register_tools(ctx) function and declare them in the manifest:

provides_tools:
  - my_platform_call
  - my_platform_list

With provides_tools declared, Hermes imports only tools.py during plugin discovery and registers the client tools in every process — CLI and TUI included — while the adapter stays deferred. Keep the package __init__.py import-light and pull the adapter in from inside register() so the eager import stays cheap. Without the field, nothing changes: the whole plugin stays deferred.

Users enable the toolset per platform like any other, e.g. hermes tools enable my_platform --platform cli, or by listing the toolset key under platform_toolsets in config.yaml. Plugin platform names are also valid --platform targets, so an inbound session on your platform can be granted its own outbound tools.

adapter.py​

from gateway.platforms._shared import extra_or_secret, get_scoped_secret, seed_extra_from_env
from gateway.platforms.base import BasePlatformAdapter, SendResult
from gateway.platforms.event import MessageEvent, MessageType
from gateway.config import Platform, PlatformConfig


class MyPlatformAdapter(BasePlatformAdapter):
    def __init__(self, config: PlatformConfig):
        super().__init__(config, Platform("my_platform"))
        # Explicit env (profile-scoped) → this profile's config.extra (YAML) → default. Under multiplexing a
        # scoped miss falls to the profile's own YAML, never to another profile's process env.
        self.token = extra_or_secret(config.extra, "token", "MY_PLATFORM_TOKEN")

    async def connect(self, *, is_reconnect: bool = False) -> bool:
        # Connect to the platform API, start listeners
        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):
        # Send message via platform API
        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(get_scoped_secret("MY_PLATFORM_TOKEN"))


def validate_config(config) -> bool:
    return bool(extra_or_secret(getattr(config, "extra", None), "token", "MY_PLATFORM_TOKEN"))


def _env_enablement() -> dict | None:
    if not get_scoped_secret("MY_PLATFORM_TOKEN", "").strip():
        return None
    return seed_extra_from_env(
        (("MY_PLATFORM_TOKEN", "token", None), ("MY_PLATFORM_CHANNEL", "channel", None)),
        home_env="MY_PLATFORM_HOME_CHANNEL",
    )


def register(ctx):
    """Plugin entry point — called by the Hermes plugin system."""
    ctx.register_platform(
        name="my_platform",
        label="My Platform",
        adapter_factory=MyPlatformAdapter,
        # PASSIVE probe — "are deps/config present right now?".  Called from
        # status displays and config loading, so it must NEVER pip-install.
        check_fn=check_requirements,
        # ACTIVE installer (optional) — only for platforms with a
        # lazy-installable SDK.  create_adapter() calls it when check_fn
        # returns False, right before the gateway connects the platform.
        # Typically wraps pm.extras.ensure_and_bind(...).  Omit it
        # and a False check_fn is a hard block.
        # ensure_deps_fn=ensure_requirements,
        validate_config=validate_config,
        required_env=["MY_PLATFORM_TOKEN"],
        install_hint="Declare my-platform-sdk in this plugin's Python dependencies, then retry hermes plugins enable my-platform",
        # Env-driven auto-configuration — seeds PlatformConfig.extra from
        # env vars before adapter construction. See "Env-Driven Auto-
        # Configuration" section below.
        env_enablement_fn=_env_enablement,
        # Cron home-channel delivery support. Lets deliver=my_platform cron
        # jobs route without editing cron/scheduler.py. See "Cron Delivery"
        # section below.
        cron_deliver_env_var="MY_PLATFORM_HOME_CHANNEL",
        # Per-platform user authorization env vars
        allowed_users_env="MY_PLATFORM_ALLOWED_USERS",
        allow_all_env="MY_PLATFORM_ALLOW_ALL_USERS",
        # Message length limit for smart chunking (0 = no limit)
        max_message_length=4000,
        # LLM guidance injected into system prompt
        platform_hint=(
            "You are chatting via My Platform. "
            "It supports markdown formatting."
        ),
        # Display
        emoji="💬",
    )

    # Optional: register platform-specific tools
    ctx.register_tool(
        name="my_platform_search",
        toolset="my_platform",
        schema={...},
        handler=my_search_handler,
    )

Configuration​

Users configure the platform in config.yaml:

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

Or via environment variables (which the adapter reads in __init__).

What the Plugin System Handles Automatically​

When you call ctx.register_platform(), the following integration points are handled for you — no core code changes needed:

Integration pointHow it works
Gateway adapter creationRegistry checked before the built-in _BUILTIN_ADAPTERS table
Config parsingPlatform._missing_() accepts any platform name
Connected platform validationRegistry validate_config() called
User authorizationallowed_users_env / allow_all_env checked; group_allowed_chats_env / allowlist_suppresses_pairing (see below)
Env-only auto-enableenv_enablement_fn seeds PlatformConfig.extra + home_channel
YAML config bridgeapply_yaml_config_fn translates config.yaml keys into env vars / extras
Cron deliverycron_deliver_env_var makes deliver=<name> work
hermes config UI entriesrequires_env / optional_env in plugin.yaml auto-populate
send engine (tools/send_message_tool.py)Routes through live gateway adapter
Webhook cross-platform deliveryRegistry checked for known platforms
/update command accessallow_update_command flag
Channel directoryPlugin platforms included in enumeration
System prompt hintsplatform_hint injected into LLM context; platform_hint_tool_notes ({tool: sentence}) appended only when the session has that tool
Message chunkingmax_message_length for smart splitting
PII redactionpii_safe flag
hermes statusLists plugin platforms alongside built-ins, one row each, using the gateway's own configured check
hermes gateway setupPlugin platforms appear in setup menu
hermes tools / hermes skillsPlugin platforms in per-platform config
Token lock (multi-profile)Use acquire_scoped_lock() in your connect()
Orphaned config warningDescriptive log when plugin is missing
Service-event authorizationtrusted_inbound=True skips user allowlists/pairing (see below)
Display defaultsdisplay_tier picks the built-in tool-progress/streaming tier
Profile clone credential stripshared_env_prefixes keeps tool-shared keys when the adapter is not in use

Service adapters: trusted_inbound, display_tier, shared_env_prefixes​

Three optional ctx.register_platform(...) / PlatformEntry fields exist for adapters that bridge a service rather than a chat network. The Home Assistant plugin is the reference consumer.

FieldType / defaultEffect
trusted_inboundbool = FalseEvery inbound event comes from the service the adapter authenticated to with its own credential; there is no human sender. The gateway's user allowlists and DM pairing do not apply to this platform. Home Assistant's event bus is the consumer. Never set it for a chat platform — it would let anyone who can message the bot drive the agent.
display_tierstr = ''One of 'high', 'medium', 'low', 'minimal': the built-in per-platform display defaults tier (tool progress, streaming, interim messages) used when the user has no display.platforms.<name> override. Empty keeps the generic plugin default. Home Assistant uses 'minimal'.
shared_env_prefixestuple[str, ...] = ()Env prefixes the platform shares with a non-channel capability, such as the same plugin's tools (Home Assistant: ('HASS_',)). Profile clones strip keys under these prefixes only when the source profile actually runs the adapter, so a profile that only uses the tools keeps its credentials.
import dataclasses

def register(ctx):
    kwargs = dict(
        name="my_service",
        label="My Service",
        adapter_factory=MyServiceAdapter,
        check_fn=check_requirements,
    )
    # Feature-detect: older cores reject unknown register_platform kwargs.
    from gateway.platform_registry import PlatformEntry
    known = {f.name for f in dataclasses.fields(PlatformEntry)}
    for key, value in (
        ("trusted_inbound", True),
        ("display_tier", "minimal"),
        ("shared_env_prefixes", ("MY_SERVICE_",)),
    ):
        if key in known:
            kwargs[key] = value
    ctx.register_platform(**kwargs)

Feature-detect these fields with dataclasses.fields(PlatformEntry) as above so the plugin stays loadable on older Hermes cores that predate them.

Authorization fields for chat platforms​

Two more optional PlatformEntry fields carry gateway authorization behaviour that core used to hard-code for its bundled platforms. Declare them and a plugin platform behaves the same way:

FieldType / defaultEffect
group_allowed_chats_envstr = ''Env var holding comma-separated group/forum chat IDs. Any message from a listed chat is authorized, even without a user_id (anonymous admins, channel posts). Telegram declares TELEGRAM_GROUP_ALLOWED_CHATS and QQ Bot declares QQ_GROUP_ALLOWED_USERS.
allowlist_suppresses_pairingbool = FalseWhen the platform's allowed_users_env (or group_allowed_chats_env) is set, unknown DM senders are ignored rather than sent a pairing code: the owner already restricted access. Most bundled chat platforms set it.
kwargs = dict(name="my_chat", label="My Chat", adapter_factory=MyChatAdapter, check_fn=check_requirements,
              allowed_users_env="MY_CHAT_ALLOWED_USERS", allow_all_env="MY_CHAT_ALLOW_ALL_USERS")
known = {f.name for f in dataclasses.fields(PlatformEntry)}
for key, value in (("group_allowed_chats_env", "MY_CHAT_GROUP_ALLOWED_CHATS"),
                   ("allowlist_suppresses_pairing", True)):
    if key in known:  # older cores reject unknown kwargs
        kwargs[key] = value
ctx.register_platform(**kwargs)

Public helpers for platform plugins​

A platform adapter needs more of core than a tool plugin does. These names are public: hermes plugins validate raises no private-import warning for them, and they keep their names and signatures across releases. Anything else under gateway.*, tools.* or hermes_cli.* is internal, and the validator lists each import of it.

NeedImport
Adapter base class, send results, media cache helpersgateway.platforms.base (public names)
SSRF-safe HTTP client: connect-time IP validation, per-redirect re-check, shared keepalive limitsfrom gateway.platforms.helpers import platform_http_client
The pieces of that client, to build your owngateway.platforms.base.ssrf_redirect_guard (httpx response hook), tools.url_safety.create_ssrf_safe_async_client (sync: create_ssrf_safe_client), tools.url_safety.is_safe_url, gateway.platforms._http_client_limits.platform_httpx_limits
Fetch a URL with an existing aiohttp.ClientSession, every redirect hop SSRF-checked (credential-bearing cross-origin hops refused, one total timeout for the chain)async with gateway.platforms.helpers.ssrf_checked_aiohttp_get(session, url, **get_kwargs) as resp: (raises ValueError on a blocked hop)
Own DM/group policy (dm_policy / group_policy / allowlists)from gateway.platforms.access_policy_mixin import OwnAccessPolicyMixin
Resolve an approval from a native buttonfrom tools.approval import resolve_gateway_approval_card (pass the card's request id; a call without one resolves nothing. Every request starts Deny-only: render through _send_exec_approval_prompt so the base send_exec_approval can lift the mark once the card shows the whole command. resolve_gateway_approval answers the oldest pending request)
Refuse a click from someone who may not answer the approval (not its requester when approvals.approvers is narrowed; not allowed /approve / /deny when the chat sets allow_admin_from)from gateway.approval_actor import approval_refusal, refusal_text: call approval_refusal(session_key, clicker_id, extra=self.config.extra, request_id=..., choice=choice) before resolving (the choice lets the chat's slash admin gate check the answer's own /approve or /deny); a non-None result is the requester's name, reply privately with refusal_text(name) and leave the approval pending. Then settle with resolve_gateway_approval_card(session_key, choice, request_id, actor_id=clicker_id, extra=self.config.extra), which rechecks the clicker against the entry it settles in the same critical section. Slash-confirm and clarify taps use slash_confirm_refusal(session_key, confirm_id, clicker_id, extra=...) and clarify_refusal(clarify_id, clicker_id, extra=...) the same way, before slash_confirm.resolve / resolve_gateway_clarify / mark_awaiting_text
Refuse a picker-card (send_model_picker / send_choice_picker) or /update prompt click from someone other than its openerfrom gateway.approval_actor import picker_refusal, update_prompt_refusal, refusal_text: picker_refusal(on_selected, clicker_id, extra=self.config.extra) with the callback the picker was sent with, or update_prompt_refusal(clicker_id, extra=self.config.extra); a non-None result is the opener's name: reply privately with refusal_text(name) and leave the card open. The opener and the chat scope's admins always pass
Native clarify buttonstools.clarify_gateway: get_pending(clarify_id) (the pending entry or None; read session_key, choices, multi_select), resolve_gateway_clarify(clarify_id, text), mark_awaiting_text(clarify_id), SKIPPED
setup_fn wizard stepshermes_cli.gateway_setup_wizard: print_setup_header(title), confirm_reconfigure(label, *env_vars), offer_home_channel(home_var, user_id, what), save_env_values(**values), prompt_csv(text, default); prompts from hermes_cli.setup (prompt, prompt_choice, prompt_yes_no); output from hermes_cli.cli_output (print_info, print_success, print_warning, print_error); .env access through hermes_cli.config (save_env_value, get_env_value)
from gateway.platforms.helpers import platform_http_client

async def connect(self) -> bool:
    # timeout=30.0 and follow_redirects=True by default; any other httpx.AsyncClient kwarg passes through.
    self._http = platform_http_client(headers={"User-Agent": "my-chat-bot/1.0"})
    ...

get_pending() returns None once the clarify was answered, timed out or cleared, so compare the result by identity with the entry your card was rendered for before acting on a click.

Read a conversation's model settings​

BasePlatformAdapter.get_session_model_settings(source) is a synchronous, read-only accessor for native settings sheets. The named consumer is the Inline platform plugin: its agent settings sheet currently reads the runner's private session-model override map and combines it with configured defaults. Together with list_session_model_picker_providers below, it supports migrating both the settings-sheet model/provider/session-key display and picker catalog off that private map read. The raw endpoint stays in core for catalog matching; the sanitized display URL is not an operational input. This does not migrate model switching: Inline still calls the private GatewayRunner._handle_model_command to apply settings-sheet model selections. No plugin source or installed copy is changed by these additive seams.

# In a BasePlatformAdapter subclass, using the SessionSource built for this conversation:
settings = self.get_session_model_settings(source)
if settings is not None:
    current_model = settings["model"]
    current_provider = settings["provider"]
    current_base_url = settings["base_url"]

The result is a fresh SessionModelSettings dictionary (type import: from gateway.platforms.session_model import SessionModelSettings) with exactly session_key, model, provider, and base_url, all strings. It is detached in both directions: editing it cannot change the conversation, and later model switches cannot change a previously returned result. An unattached adapter or rejected profile route returns None. An attached adapter with no session still returns configured defaults.

For each setting, a nonempty in-memory session override (or, if absent, the persisted pick) wins over the channel default (model/provider), then the owning profile's configured model settings. Channel model and provider defaults apply independently, including provider-only overrides. The display base_url falls back independently to the global configured endpoint even when the selected provider changes; it can belong to a different provider. Do not present that provider/endpoint pair as routing information. Missing model and endpoint are ""; missing provider is "openrouter", as in Inline's existing display. Both a flat model: name and the mapping form (default, or legacy model) are supported. The source's existing routing identity, or the adapter's canonical routing resolver on a copy of the source, selects the profile; the caller's source is not modified.

This is a settings display snapshot, not a credential or live-inference introspection API. It does not allocate a session or agent, load a persisted override into memory, resolve provider credentials, contact a provider, or change any settings. A persisted /model choice not yet loaded after a restart is read from the session store; a transient runtime fallback is not reported. Only the four named fields are returned: keys, tokens, credential pools and headers are never copied. Endpoint userinfo, query and fragment are omitted (unparseable URLs become ""); the endpoint path is retained, so do not put credentials in endpoint paths. Do not use this display URL to construct provider requests. This is an in-process platform adapter API, not a hosted-plugin RPC. It adds no setters or reasoning API, and leaves the legacy override view and all existing model-switch behavior unchanged.

List a conversation's model picker providers​

BasePlatformAdapter.list_session_model_picker_providers(source, *, max_models=100, include_moa=True) is the synchronous catalog counterpart. It shares the display accessor's source canonicalization, owning-profile config scope, session > channel > config precedence, and persisted-pick fallback. It returns None without a runner or for a rejected route; an attached adapter with no session can still list configured providers. It does not create a session or agent, rehydrate an override, or change conversation settings.

Core passes the raw effective endpoint, current model/provider, configured providers, custom providers (the same merged view the built-in /model picker uses), and model_catalog.excluded_providers to its picker catalog implementation. The owning profile scope remains bound throughout the catalog call, including when this method is invoked through asyncio.to_thread. Do not feed the display URL into discovery. Custom endpoint live probes are disabled: those probes can save discovered models to config. Configured and cached custom model lists still participate, including raw endpoint matching. Catalogs use the built-in /model picker's non-blocking path: remote catalogs come from the disk cache and stale ones refresh in the background under the owning profile. File reads and a short local LM Studio probe (1.5 s timeout) remain, so run this method off the event loop. max_models is the catalog's per-provider hint, not a hard total cap; curated/aggregator providers may be uncapped. include_moa includes the synthetic MoA row.

The return value is a detached list of fresh SessionModelPickerProvider dictionaries (type import: from gateway.platforms.session_model import SessionModelPickerProvider), with exactly these allowlisted fields:

FieldTypeInline use
slugstrModel selection identity and provider grouping
namestrProvider buttons and model-page heading
modelslist[str]Model options; list copied independently
total_modelsintProvider/group counts and pagination text
is_currentboolCurrent-provider marker

No endpoint fields, API keys, tokens, credential pools, headers, or unlisted future metadata are returned. Editing rows or model lists cannot change catalog/config/session objects, and later catalog changes cannot change an earlier result. This is presentation data, not runtime credentials or a setter. Catalog errors propagate for the adapter to handle.

import asyncio

# Feature-detect for older Hermes: leave model selection unavailable (or retain an
# explicitly supported legacy implementation) when the seam is absent. Never replace
# the raw endpoint with get_session_model_settings()["base_url"].
list_providers = getattr(self, "list_session_model_picker_providers", None)
providers = None
if callable(list_providers):
    providers = await asyncio.to_thread(list_providers, source, max_models=100, include_moa=True)

The seams enable an Inline migration; they do not deploy it. Updating a catalog entry does not upgrade an already-installed plugin. Inline must adopt the methods and users must separately update their installed plugin; its private model-switch command remains outside this read-only contract.

Standalone send-path extensions​

A standalone platform can participate in host-driven outbound delivery through direct hermes send --to ... and cron deliver=platform:... by declaring send behavior on the same PlatformEntry created by ctx.register_platform(). send_message is intentionally not an agent-callable model tool; plugins must not register an equivalent model surface that lets the agent initiate outbound messages on its own.

async def _send_request(args, chat_id, platform_name, pconfig):
    # `args` contains the host-driven send request fields.
    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, optional thread_id)
    return None                 # continue to channel-directory resolution


def _validate_address(address):
    # True accepts; False rejects; a string rejects with that diagnostic.
    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,
        # May be a regular function or async def. Hermes awaits any awaitable
        # result, including callable objects and functools.partial wrappers.
        send_message_handler=_send_request,
        # Prefer this lower-level hook when cron must send from a process
        # without the live gateway.
        standalone_sender_fn=_standalone_send,
    )

Target resolution is shared across all three outbound surfaces. Parser output is normalized first and channel-directory IDs are trusted. A plugin parser must explicitly accept native target syntax; unresolved strings are never passed through opaquely. Unknown platforms and validator failures return a diagnostic instead of silently attempting delivery. Plugin force-reload/profile transitions unregister owned entries, so parsers and handlers cannot leak into the next profile.

Env-Driven Auto-Configuration​

Most users set up a platform by dropping env vars into ~/.hermes/.env rather than editing config.yaml. The env_enablement_fn hook lets your plugin pick those env vars up before the adapter is constructed, so hermes gateway status, get_connected_platforms(), and cron delivery see the correct state without instantiating the platform SDK.

Read env through gateway.platforms._shared.get_scoped_secret — never os.getenv. Under gateway.multiplex_profiles the process env holds the DEFAULT profile's values; a secondary profile's .env exists only in its secret scope, and a raw read would enable your platform for the wrong profile with the wrong credentials. seed_extra_from_env builds the seed dict from a (ENV_VAR, extra_key, conv) table through that reader.

from gateway.platforms._shared import get_scoped_secret, seed_extra_from_env


def _env_enablement() -> dict | None:
    """Seed PlatformConfig.extra from env vars.

    Called by the platform registry during load_gateway_config().
    Return None when the platform isn't minimally configured — the
    caller then skips auto-enabling. Return a dict to seed extras.
    """
    token = get_scoped_secret("MY_PLATFORM_TOKEN", "").strip()
    channel = get_scoped_secret("MY_PLATFORM_CHANNEL", "").strip()
    if not (token and channel):
        return None
    # (ENV_VAR, extra_key, conv): blank values are skipped, a conv raising
    # ValueError drops the key. home_env seeds the special 'home_channel'
    # dict ({chat_id, name} from MY_PLATFORM_HOME_CHANNEL[_NAME], name
    # defaults to "Home"), which the core hook lifts into a HomeChannel.
    return {"token": token, "channel": channel, **seed_extra_from_env(
        (("MY_PLATFORM_PORT", "port", int), ("MY_PLATFORM_NICK", "nick", None)),
        home_env="MY_PLATFORM_HOME_CHANNEL",
    )}


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

YAML→env Config Bridge​

Some users prefer setting config.yaml keys (my_platform.require_mention, my_platform.allowed_channels, etc.) over env vars. The apply_yaml_config_fn hook lets your plugin own this translation instead of forcing core gateway/config.py to know your platform's YAML schema.

from gateway.platforms._shared import apply_yaml_bridge

# (yaml key, ENV_VAR, kind). "lower" bridges whenever the key is present
# (booleans become "true"/"false"); "str" skips null/blank; "csv" comma-joins
# lists; "json" dumps the value.
_YAML_BRIDGE = (
    ("require_mention", "MY_PLATFORM_REQUIRE_MENTION", "lower"),
    ("allowed_channels", "MY_PLATFORM_ALLOWED_CHANNELS", "csv"),
)


def _apply_yaml_config(yaml_cfg: dict, platform_cfg: dict) -> dict | None:
    """Translate config.yaml `my_platform:` keys into env vars / extras.

    yaml_cfg     — the full top-level parsed config.yaml dict
    platform_cfg — the platform's own sub-dict (yaml_cfg.get("my_platform", {}))

    apply_yaml_bridge writes each env var only when it is unset (env > YAML)
    and NEVER while a multiplexed secondary profile's config is loading — a
    write there would pin that profile's policy process-wide. It returns the
    same values for PlatformConfig.extra, so read `extra` first in the adapter
    and fall back to the env var (`_shared.extra_or_secret`).
    """
    return apply_yaml_bridge(platform_cfg, _YAML_BRIDGE)

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

The hook is invoked during load_gateway_config() after the generic shared-key loop (which handles common keys like unauthorized_dm_behavior, notice_delivery, reply_prefix, require_mention, etc.) and before _apply_env_overrides(), so your plugin only needs to bridge platform-specific keys.

Exceptions raised by the hook are swallowed and logged at debug level — a misbehaving plugin never aborts gateway config load.

Cron Delivery​

To let deliver=my_platform cron jobs route to a configured home channel, set cron_deliver_env_var to the env var name that holds the default chat/room/channel ID:

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

The scheduler reads this env var when resolving the home target for deliver=my_platform jobs, and also treats the platform as a valid cron target in _KNOWN_DELIVERY_PLATFORMS-style checks. If your env_enablement_fn seeds a home_channel dict (see above), that takes precedence — cron_deliver_env_var is the fallback for cron jobs that run before env seeding.

Out-of-process cron delivery​

cron_deliver_env_var makes your platform a recognized deliver= target. To make the actual send succeed when the cron job runs in a separate process from the gateway (i.e., hermes cron run separate from hermes gateway), register a standalone_sender_fn:

async def _standalone_send(
    pconfig,
    chat_id,
    message,
    *,
    thread_id=None,
    media_files=None,
    force_document=False,
):
    """Open an ephemeral connection / acquire a fresh token, send, and close."""
    # ... open connection, send message, return result ...
    return {"success": True, "message_id": "..."}
    # or {"error": "..."}

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

Why this hook is necessary: the live adapter is only reachable through _gateway_runner_ref(), which returns None outside the gateway process, so without standalone_sender_fn the cron-side send fails with No live adapter for platform '<name>'. The bundled platforms use this same hook: Telegram's out-of-process sender (hermes send, the send_message tool, standalone cron delivery) is plugins/platforms/telegram/standalone.py::standalone_send and Signal's is plugins/platforms/signal/standalone.py; core reaches them only through the platform registry.

The function receives the same pconfig and chat_id that the live adapter would, plus optional thread_id, media_files, and force_document keyword arguments. Optional keywords added later are forwarded only to a sender that declares them (by name or through **kwargs), so a sender written to the signature above keeps working: notify=True (cron's cron.delivery.notify: this delivery should ring even under a quiet notification mode) is one; Telegram's sender takes it. Returning {"success": True, "message_id": ...} is treated as a successful delivery; returning {"error": "..."} surfaces the message in cron's delivery_errors. Exceptions raised inside the function are caught by the dispatcher and reported as Plugin standalone send failed: <reason>. Reference implementations live in plugins/platforms/{irc,teams,google_chat}/adapter.py.

Surfacing Env Vars in hermes config​

hermes_cli/config.py scans plugins/platforms/*/plugin.yaml at import time and auto-populates OPTIONAL_ENV_VARS from requires_env and (optional) optional_env blocks. Use the rich-dict form to contribute proper descriptions, prompts, password flags, and URLs — the CLI setup UI picks them up for free.

# plugins/platforms/my_platform/plugin.yaml
name: my_platform-platform
label: My Platform
kind: platform
version: 1.0.0
description: >
  My Platform gateway adapter for Hermes Agent.
author: Your Name
requires_env:
  - name: MY_PLATFORM_TOKEN
    description: "Bot API token from the My Platform console"
    prompt: "My Platform bot token"
    url: "https://my-platform.example.com/bots"
    password: true
  - name: MY_PLATFORM_CHANNEL
    description: "Channel to join (e.g. #hermes)"
    prompt: "Channel"
    password: false
optional_env:
  - name: MY_PLATFORM_HOME_CHANNEL
    description: "Default channel for cron delivery (defaults to MY_PLATFORM_CHANNEL)"
    prompt: "Home channel (or empty)"
    password: false
  - name: MY_PLATFORM_ALLOWED_USERS
    description: "Comma-separated user IDs allowed to talk to the bot"
    prompt: "Allowed users (comma-separated)"
    password: false

Supported dict keys: name (required), description, prompt, url, password (bool; auto-detected from *_TOKEN / *_SECRET / *_KEY / *_PASSWORD / *_JSON suffix when omitted), category (defaults to "messaging").

Bare-string entries (- MY_PLATFORM_TOKEN) still work — they get a generic description auto-derived from the plugin's label. If a hardcoded entry for the same var already exists in OPTIONAL_ENV_VARS, it wins (back-compat); the plugin.yaml form acts as the fallback.

Platform-Specific Slow-LLM UX​

Some platforms have constraints that change how a slow LLM response should be presented:

  • LINE issues a single-use reply token that expires roughly 60 seconds after the inbound event. Replying with that token is free; falling back to the metered Push API is not. If the LLM hasn't finished by the deadline, the choice is "burn paid Push quota" or "do something cleverer with the reply token before it expires."
  • WhatsApp marks a session inactive after 24h, after which only template messages are accepted.
  • SMS has no concept of typing indicators or progressive updates — long responses just look like the bot is offline.

These are real constraints the base BasePlatformAdapter can't anticipate. The plugin surface intentionally leaves the room for an adapter to layer platform-specific UX on top of the base typing loop without expanding the kwarg list.

Pattern: subclass _keep_typing to layer mid-flight UX​

BasePlatformAdapter._keep_typing is the typing-indicator heartbeat — it runs as a background task while the LLM is generating, and is cancelled when the response is delivered. To layer a platform-specific behavior at a threshold (e.g. send a "still thinking" bubble at 45s), override _keep_typing in your adapter, schedule your own task alongside super()._keep_typing(), and tear it down in 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
            # Platform-specific work here — for LINE, send a Template
            # Buttons "Get answer" bubble using the cached reply token
            # so the user can fetch the cached response later via a
            # fresh (free) reply token from the postback callback.
            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

Key points:

  • Always await super()._keep_typing(...). The typing heartbeat is independently useful — don't replace it, layer on top of it.
  • Tear down the side task in finally. When the LLM finishes (or /stop cancels the run), the gateway cancels the typing task. Your side task must observe that cancellation too, otherwise it lingers and may fire after the response was already delivered.
  • Pair with interrupt_session_activity to resolve any orphan UX state when the user issues /stop. For LINE, this means transitioning the postback cache entry from PENDING to ERROR so the persistent "Get answer" button delivers a "Run was interrupted" message instead of looping.

Pattern: subclass send to route through a cache instead of sending immediately​

If your slow-response UX caches the response for later retrieval (LINE's postback flow), your send override needs to recognize three modes:

  1. Pending postback active for this chat → cache the response under the request_id, don't send anything visible.
  2. System busy-ack (⚡ Interrupting, ⏳ Queued, ⏩ Steered) → bypass the cache and send visibly so the user sees the gateway's response to their input.
  3. Normal response → send via reply-token-or-push as usual.
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)

_SYSTEM_BYPASS_PREFIXES are the gateway's own busy-acknowledgment prefixes (⚡, ⏳, ⏩, 💾). Always let those through visibly, regardless of cached UX state.

When this pattern is appropriate​

Use the typing-loop override approach when:

  • The platform's outbound API has a hard time-window constraint (single-use reply token, expiring sticky session, etc.) AND
  • A visible mid-flight bubble is acceptable UX on that platform.

Use the simpler slow_response_threshold = 0 always-Push path when:

  • The platform doesn't have a meaningful free vs. paid distinction, OR
  • The user community prefers "loading… loading… DONE" silence-then-response over an interactive intermediate bubble.

LINE supports both: the threshold defaults to 45s for free postback fetch, and LINE_SLOW_RESPONSE_THRESHOLD=0 reverts to "always Push fallback."

Reference Implementation​

See plugins/platforms/line/adapter.py for the full LINE postback implementation — a RequestCache state machine (PENDING → READY → DELIVERED, plus ERROR for /stop), a _keep_typing override that fires the Template Buttons bubble at threshold, a send override that routes through the cache, and an interrupt_session_activity override that resolves orphan PENDING entries.

Reference Implementations (Plugin Path)​

See plugins/platforms/irc/ in the repo for a complete working example — a full async IRC adapter with zero external dependencies. plugins/platforms/teams/ covers Bot Framework / Adaptive Cards, plugins/platforms/google_chat/ covers OAuth-based REST APIs, and plugins/platforms/line/ covers webhook-driven Messaging APIs with platform-specific slow-LLM UX.

Step-by-Step Checklist (Built-in Path)​

note

This checklist is for adding a platform directly to the Hermes core codebase — typically done by core contributors for officially supported platforms. Community/third-party platforms should use the Plugin Path above.

1. Platform Enum​

Add your platform to the Platform enum in gateway/config.py:

class Platform(Enum):
    # ... existing platforms ...
    NEWPLAT = "newplat"

2. Adapter File​

Create plugins/platforms/newplat/adapter.py:

from gateway.config import Platform, PlatformConfig
from gateway.platforms._shared import extra_or_secret
from gateway.platforms.base import BasePlatformAdapter, SendResult
from gateway.platforms.event import MessageEvent, MessageType

def check_newplat_requirements() -> bool:
    """Return True if dependencies are available."""
    return SOME_SDK_AVAILABLE

class NewPlatAdapter(BasePlatformAdapter):
    def __init__(self, config: PlatformConfig):
        super().__init__(config, Platform.NEWPLAT)
        # Read config from config.extra dict
        extra = config.extra or {}
        self._api_key = extra_or_secret(extra, "api_key", "NEWPLAT_API_KEY")

    async def connect(self, *, is_reconnect: bool = False) -> bool:
        # Set up connection, start 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):
        # Send message via platform API
        return SendResult(success=True, message_id="...")

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

For inbound messages, build a MessageEvent and call self.handle_message(event):

source = self.build_source(
    chat_id=chat_id,
    chat_name=name,
    chat_type="dm",  # or "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. Gateway Config (gateway/config.py)​

Three touchpoints:

  1. get_connected_platforms() — Add a check for your platform's required credentials
  2. load_gateway_config() — Add token env map entry: Platform.NEWPLAT: "NEWPLAT_TOKEN"
  3. _apply_env_overrides() — Map all NEWPLAT_* env vars to config

4. Gateway Runner (gateway/run.py + gateway/run_*.py siblings)​

Six touchpoints:

  1. _BUILTIN_ADAPTERS table (gateway/run.py) — Add a Platform.NEWPLAT: (module, class, check_fn, error_msg) entry; _instantiate_adapter() (gateway/run_adapters.py) consults the plugin registry, then this table — there is no elif chain to extend. The _create_adapter() wrapper binds every successful adapter to its gateway runner.
  2. _is_user_authorized() allowed_users map — Platform.NEWPLAT: "NEWPLAT_ALLOWED_USERS"
  3. _is_user_authorized() allow_all map — Platform.NEWPLAT: "NEWPLAT_ALLOW_ALL_USERS"
  4. Startup access-policy check (gateway/run_startup.py) — Add "NEWPLAT" to _ALLOWLIST_ENV_PLATFORMS (derives both NEWPLAT_ALLOWED_USERS and NEWPLAT_ALLOW_ALL_USERS)
  5. Startup _BUILTIN_ALLOW_ALL_VARS (gateway/run_startup.py) — derived from the same _ALLOWLIST_ENV_PLATFORMS tuple; nothing extra to add
  6. _UPDATE_ALLOWED_PLATFORMS frozenset — Add Platform.NEWPLAT

5. Cross-Platform Delivery​

  1. gateway/platforms/webhook.py — Add "newplat" to the delivery type tuple
  2. cron/scheduler_delivery.py — Add to _KNOWN_DELIVERY_PLATFORMS frozenset and _deliver_result() platform map

6. CLI Integration​

  1. hermes_cli/config.py — Add all NEWPLAT_* vars to _EXTRA_ENV_KEYS
  2. hermes_cli/gateway.py — Add entry to _PLATFORMS list with key, label, emoji, token_var, setup_instructions, and vars
  3. hermes_cli/platforms.py — Add PlatformInfo entry with label and default_toolset (used by skills_config and tools_config TUIs)
  4. hermes_cli/setup.py — Add _setup_newplat() function (can delegate to gateway.py) and add tuple to the messaging platforms list
  5. hermes_cli/status.py — Add platform detection entry: "NewPlat": ("NEWPLAT_TOKEN", "NEWPLAT_HOME_CHANNEL")
  6. hermes_cli/dump.py — Add "newplat": "NEWPLAT_TOKEN" to platform detection dict

7. Tools​

  1. tools/send_message_tool.py — Add "newplat": Platform.NEWPLAT to platform map
  2. tools/cronjob_tools.py — Add newplat to the delivery target description string

8. Toolsets​

  1. toolsets.py — Add "hermes-newplat" toolset definition with _HERMES_CORE_TOOLS
  2. toolsets.py — Add "hermes-newplat" to the "hermes-gateway" includes list

9. Optional: Platform Hints​

platform_hint= on ctx.register_platform — If your platform has specific rendering limitations (no markdown, message length limits, etc.), declare a hint on its registry entry (keep the text in a light sibling such as adapter_hint.py). It is injected into the system prompt; agent/prompt_builder.py::PLATFORM_HINTS holds only core surfaces:

PLATFORM_HINT = (
    "You are chatting via NewPlat. It supports markdown formatting "
    "but has a 4000-character message limit."
)

Not all platforms need hints — only add one if the agent's behavior should differ.

10. Tests​

Create tests/gateway/test_newplat.py covering:

  • Adapter construction from config
  • Message event building
  • Send method (mock the external API)
  • Platform-specific features (encryption, routing, etc.)

11. Documentation​

FileWhat to add
website/docs/user-guide/messaging/newplat.mdFull platform setup page
website/docs/user-guide/messaging/index.mdPlatform comparison table, architecture diagram, toolsets table, security section, next-steps link
website/docs/reference/environment-variables.mdAll NEWPLAT_* env vars
website/docs/reference/toolsets-reference.mdhermes-newplat toolset
website/docs/integrations/index.mdPlatform link
website/sidebars.tsSidebar entry for the docs page
website/docs/developer-guide/architecture.mdAdapter count + listing
website/docs/developer-guide/gateway-internals.mdAdapter file listing

Parity Audit​

Before marking a new platform PR as complete, run a parity audit against an established platform:

# Find every .py file mentioning the reference platform
search_files "bluebubbles" output_mode="files_only" file_glob="*.py"

# Find every .py file mentioning the new platform
search_files "newplat" output_mode="files_only" file_glob="*.py"

# Any file in the first set but not the second is a potential gap

Repeat for .md and .ts files. Investigate each gap — is it a platform enumeration (needs updating) or a platform-specific reference (skip)?

Common Patterns​

Long-Poll Adapters​

If your adapter uses long-polling (like Telegram or Weixin), use a polling loop task:

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))

Callback/Webhook Adapters​

If the platform pushes messages to your endpoint (like WeCom Callback), run an HTTP server:

async def connect(self):
    self._app = web.Application()
    self._app.router.add_post("/callback", self._handle_callback)
    # ... start aiohttp server
    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")  # Acknowledge immediately

For platforms with tight response deadlines (e.g., WeCom's 5-second limit), always acknowledge immediately and deliver the agent's reply proactively via API later. Agent sessions run 3–30 minutes — inline replies within a callback response window are not feasible.

Inbound Deduplication​

Platforms redeliver: websocket resumes replay recent events, webhooks retry, and an unacknowledged poll batch comes back. Drop repeats with the shared helper, keyed on the platform's message ID:

from gateway.platforms.helpers import MessageDeduplicator

self._dedup = MessageDeduplicator(ttl_seconds=600)  # in __init__

if self._dedup.is_duplicate(msg_id):  # in the inbound handler
    return

When the gateway's reconnect watcher replaces a failed adapter with a new instance, it copies every MessageDeduplicator attribute's live IDs from the old instance to the new one, so a replay right after the reconnect is still dropped. A cache kept in another structure (a plain dict or set) starts empty on the new instance.

Token Locks​

If the adapter holds a persistent connection with a unique credential, add a scoped lock to prevent two profiles from using the same credential:

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
    # ... connect

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

Reference Implementations​

AdapterPatternComplexityGood reference for
plugins/platforms/bluebubbles/adapter.pyREST + webhookMediumSimple REST API integration
plugins/platforms/weixin/adapter.pyLong-poll + CDNHighMedia handling, encryption
plugins/platforms/wecom/callback_adapter.pyCallback/webhookMediumHTTP server, AES crypto, multi-app
plugins/platforms/irc/adapter.pyLong-poll + IRC protocolHighFull-featured plugin adapter with scoped token lock