hermes-agent/website/docs/developer-guide/adding-platform-adapters.md

sidebar_position

sidebar_position
9

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 Adding a platform adapter touches 20+ files across code, config, and docs. Use this guide as a checklist — the adapter file itself is typically only 40% of the work. :::

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.

Step-by-Step Checklist

1. Platform Enum

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

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

2. Adapter File

Create gateway/platforms/newplat.py:

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

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.get("api_key") or os.getenv("NEWPLAT_API_KEY", "")

    async def connect(self) -> 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)

Five touchpoints:

1. _create_adapter() — Add an elif platform == Platform.NEWPLAT: branch
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. Early env check _any_allowlist tuple — Add "NEWPLAT_ALLOWED_USERS"
5. Early env check _allow_all tuple — Add "NEWPLAT_ALLOW_ALL_USERS"
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.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

agent/prompt_builder.py — If your platform has specific rendering limitations (no markdown, message length limits, etc.), add an entry to the _PLATFORM_HINTS dict. This injects platform-specific guidance into the system prompt:

_PLATFORM_HINTS = {
    # ...
    "newplat": (
        "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

File	What to add
website/docs/user-guide/messaging/newplat.md	Full platform setup page
website/docs/user-guide/messaging/index.md	Platform comparison table, architecture diagram, toolsets table, security section, next-steps link
website/docs/reference/environment-variables.md	All NEWPLAT_* env vars
website/docs/reference/toolsets-reference.md	hermes-newplat toolset
website/docs/integrations/index.md	Platform link
website/sidebars.ts	Sidebar entry for the docs page
website/docs/developer-guide/architecture.md	Adapter count + listing
website/docs/developer-guide/gateway-internals.md	Adapter 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.

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):
    if not acquire_scoped_lock("newplat", self._token):
        logger.error("Token already in use by another profile")
        return False
    # ... connect

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

Reference Implementations

Adapter	Pattern	Complexity	Good reference for
bluebubbles.py	REST + webhook	Medium	Simple REST API integration
weixin.py	Long-poll + CDN	High	Media handling, encryption
wecom_callback.py	Callback/webhook	Medium	HTTP server, AES crypto, multi-app
telegram.py	Long-poll + Bot API	High	Full-featured adapter with groups, threads