adding-platform-adapters

anchor alias

anchor alias

anchor alias

플랫폼 어댑터 추가

이 가이드에서는 Hermes 게이트웨이에 새로운 메시징 플랫폼을 추가하는 방법을 다룹니다. 플랫폼 어댑터는 Hermes를 외부 메시징 서비스(Telegram, Discord, WeCom 등)에 연결하므로 사용자는 해당 서비스를 통해 에이전트와 상호 작용할 수 있습니다.

팁

플랫폼을 추가하는 방법에는 두 가지가 있습니다.

• 플러그인(커뮤니티/타사에 권장): 플러그인 디렉터리를 ~/.hermes/plugins/에 놓습니다. 핵심 코드 변경이 필요하지 않습니다. 아래의 플러그인 경로를 참조하세요.
• 내장: 코드, 구성, 문서 전체에서 20개 이상의 파일을 수정합니다. 아래의 내장 체크리스트를 사용하세요.

아키텍처 개요

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

모든 어댑터는 gateway/platforms/base.py에서 BasePlatformAdapter을 확장하고 다음을 구현합니다.

• connect() — 연결 설정(WebSocket, 롱 폴, HTTP 서버 등) (추상)
• disconnect() — 완전 종료 (추상)
• send() — 채팅에 문자 메시지 보내기 (요약)
• send_typing() — 입력 표시기 표시(선택적 재정의)
• get_chat_info() — 채팅 메타데이터 반환(선택적 재정의)

인바운드 메시지는 어댑터에 의해 수신되고 기본 클래스가 게이트웨이 실행기로 라우팅되는 self.handle_message(event)을 통해 전달됩니다.

플러그인 경로(권장)

플러그인 시스템을 사용하면 핵심 Hermes 코드를 수정하지 않고도 플랫폼 어댑터를 추가할 수 있습니다. 플러그인은 두 개의 파일이 있는 디렉터리입니다.

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

플러그인.yaml

플러그인 메타데이터. requires_env 및 optional_env 블록은 hermes config UI 항목을 자동으로 채웁니다(아래 환경 변수 표시 참조).

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

어댑터.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) -> 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(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):
    """Plugin entry point — called by the Hermes plugin system."""
    ctx.register_platform(
        name="my_platform",
        label="My Platform",
        adapter_factory=lambda cfg: MyPlatformAdapter(cfg),
        check_fn=check_requirements,
        validate_config=validate_config,
        required_env=["MY_PLATFORM_TOKEN"],
        install_hint="pip install my-platform-sdk",
        # 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,
    )

구성

사용자는 config.yaml에서 플랫폼을 구성합니다.

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

또는 환경 변수(어댑터가 __init__에서 읽음)를 통해.

플러그인 시스템이 자동으로 처리하는 것

ctx.register_platform()을 호출하면 다음 통합 지점이 자동으로 처리됩니다. 핵심 코드를 변경할 필요가 없습니다.

통합 포인트	작동 원리
게이트웨이 어댑터 생성	내장 if/elif 체인 이전에 레지스트리 검사
구성 파싱	Platform._missing_()은 모든 플랫폼 이름을 허용합니다.
연결된 플랫폼 검증	레지스트리 validate_config() 호출됨
사용자 인증	allowed_users_env / allow_all_env 확인됨
환경 전용 자동 활성화	env_enablement_fn 씨앗 PlatformConfig.extra + home_channel
YAML 구성 브리지	apply_yaml_config_fn은 config.yaml 키를 환경 변수/추가 항목으로 변환합니다.
크론 전달	cron_deliver_env_var은 deliver=<name>을 작동시킵니다.
hermes config UI 항목	requires_env / optional_env(plugin.yaml) 자동 입력
send_message 도구	라이브 게이트웨이 어댑터를 통한 라우팅
웹훅 크로스 플랫폼 전달	알려진 플랫폼에 대해 레지스트리를 확인했습니다.
/update 명령 액세스	allow_update_command 플래그
채널 디렉토리	열거에 포함된 플러그인 플랫폼
시스템 프롬프트 힌트	platform_hint LLM 컨텍스트에 삽입됨
메시지 청킹	스마트 분할용 max_message_length
PII 수정	pii_safe 플래그
hermes status	(plugin) 태그가 있는 플러그인 플랫폼을 표시합니다.
hermes gateway setup	플러그인 플랫폼이 설정 메뉴에 나타납니다.
hermes tools / hermes skills	플랫폼별 구성의 플러그인 플랫폼
토큰 잠금(다중 프로필)	connect()에서 acquire_scoped_lock()을 사용하세요.
분리된 구성 경고	플러그인이 누락된 경우의 설명 로그

환경 기반 자동 구성

대부분의 사용자는 config.yaml을 편집하는 대신 환경 변수를 ~/.hermes/.env에 삭제하여 플랫폼을 설정합니다. env_enablement_fn 후크를 사용하면 플러그인이 어댑터가 구성되기 전에 해당 환경 변수를 선택할 수 있으므로 hermes gateway status, get_connected_platforms() 및 cron 전달은 플랫폼 SDK를 인스턴스화하지 않고도 올바른 상태를 볼 수 있습니다.

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.

    The special 'home_channel' key is extracted and becomes a proper
    HomeChannel dataclass on the PlatformConfig; every other key is
    merged into 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,
        #... other fields
    )

YAML→env 구성 브리지

일부 사용자는 환경 변수보다 config.yaml 키(my_platform.require_mention, my_platform.allowed_channels 등) 설정을 선호합니다. apply_yaml_config_fn 후크를 사용하면 핵심 gateway/config.py이 플랫폼의 YAML 스키마를 알도록 강제하는 대신 플러그인이 이 번역을 소유할 수 있습니다.

import os

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", {}))

    May mutate os.environ directly (use `not os.getenv(...)` guards to
    preserve env > YAML precedence) and/or return a dict to merge into
    PlatformConfig.extra. Return None or {} for no 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  # nothing extra to merge into PlatformConfig.extra

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

후크는 일반 공유 키 루프(unauthorized_dm_behavior, notice_delivery, reply_prefix, require_mention 등과 같은 공통 키를 처리함) 이후 load_gateway_config() 동안 및 그 이전에 호출됩니다. _apply_env_재정의s(), 따라서 플러그인은 플랫폼별 키만 연결하면 됩니다.

후크에 의해 발생한 예외는 디버그 수준에서 무시되고 기록됩니다. 오작동하는 플러그인은 게이트웨이 구성 로드를 중단하지 않습니다.

크론 전달

deliver=my_platform cron 작업이 구성된 홈 채널로 라우팅되도록 하려면 cron_deliver_env_var을 기본 채팅/방/채널 ID를 보유하는 환경 변수 이름으로 설정하세요.

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

스케줄러는 deliver=my_platform 작업에 대한 홈 대상을 확인할 때 이 환경 변수를 읽고 _KNOWN_DELIVERY_PLATFORMS 스타일 검사에서 플랫폼을 유효한 크론 대상으로 처리합니다. env_enablement_fn가 home_channel 딕셔너리(위 참조)를 시드하는 경우 해당 딕셔너리가 우선합니다. — cron_deliver_env_var은 환경 시드 전에 실행되는 크론 작업에 대한 대체입니다.

프로세스 외부 크론 전달

cron_deliver_env_var은 사용자의 플랫폼을 인식된 deliver= 대상으로 만듭니다. cron 작업이 게이트웨이와 별도의 프로세스(즉, hermes cron run와 별도의 hermes gateway)에서 실행될 때 실제 전송이 성공하도록 하려면 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,
)

이 후크가 필요한 이유: 내장 플랫폼(Telegram, Discord, Slack 등)은 tools/send_message_tool.py에 직접 REST 도우미를 제공하므로 cron은 동일한 프로세스에서 게이트웨이를 유지하지 않고도 전달할 수 있습니다. 플러그인 플랫폼은 역사적으로 게이트웨이 프로세스 외부에서 None을 반환하는 _gateway_runner_ref()에 의존했기 때문에 standalone_sender_fn이 없으면 cron 측 전송이 No live adapter for platform '<name>'과 함께 실패합니다.

이 함수는 라이브 어댑터와 동일한 pconfig 및 chat_id과 선택적인 thread_id, media_files 및 force_document 키워드 인수를 받습니다. {"success": True, "message_id":...} 반환은 성공적인 배송으로 처리됩니다. {"error": "..."}을 반환하면 cron의 delivery_errors에 메시지가 표시됩니다. 함수 내부에서 발생한 예외는 디스패처에 의해 포착되고 Plugin standalone send failed: <reason>로 보고됩니다. 참조 구현은 plugins/platforms/{irc,teams,google_chat}/adapter.py에 있습니다.

hermes config에 환경 변수 표시

hermes_cli/config.py은 가져오기 시 plugins/platforms/*/plugin.yaml을 스캔하고 requires_env 및 (선택 사항) optional_env 블록에서 OPTIONAL_ENV_VARS을 자동으로 채웁니다. rich-dict 양식을 사용하여 적절한 설명, 프롬프트, 비밀번호 플래그 및 URL을 제공하세요. CLI 설정 UI가 이를 무료로 선택합니다.

# 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

지원되는 사전 키: name (필수), description, prompt, url, password(부울; 다음에서 자동 감지됨) *_TOKEN / *_SECRET / *_KEY / *_PASSWORD / *_JSON 생략 시 접미사), category(기본값: "messaging").

단순 문자열 항목(- MY_PLATFORM_TOKEN)은 여전히 작동합니다. 플러그인의 label에서 자동 파생된 일반 설명을 가져옵니다. 동일한 var에 대해 하드코딩된 항목이 이미 OPTIONAL_ENV_VARS에 존재하는 경우 해당 항목이 승리합니다(역호환). 플러그인.yaml 형식이 대체 역할을 합니다.

플랫폼별 Slow-LLM UX

일부 플랫폼에는 느린 LLM 응답이 표시되는 방식을 변경하는 제약이 있습니다.

• LINE은 인바운드 이벤트 후 약 60초 후에 만료되는 일회용 응답 토큰을 발행합니다. 해당 토큰으로 답장하는 것은 무료입니다. 측정된 Push API로 돌아가는 것은 아닙니다. LLM이 기한까지 완료되지 않은 경우 "유료 푸시 할당량 소각" 또는 "만료되기 전에 응답 토큰을 사용하여 더 영리한 작업 수행" 중에서 선택할 수 있습니다.
• WhatsApp은 24시간 후에 세션을 비활성 상태로 표시하고 그 후에는 템플릿 메시지만 허용됩니다.
• SMS에는 입력 표시 또는 점진적인 업데이트 개념이 없습니다. 긴 응답은 봇이 오프라인인 것처럼 보입니다.

이는 BasePlatformAdapter 베이스가 예상할 수 없는 실제 제약 조건입니다. 플러그인 표면은 kwarg 목록을 확장하지 않고 기본 타이핑 루프 위에 어댑터가 플랫폼별 UX를 계층화할 수 있는 공간을 의도적으로 남겨둡니다.

패턴: 비행 중 UX 레이어에 대한 서브클래스 _keep_typing

BasePlatformAdapter._keep_typing은 입력 표시기 하트비트입니다. 이는 LLM이 생성되는 동안 백그라운드 작업으로 실행되고 응답이 전달되면 취소됩니다. 임계값에서 플랫폼별 동작을 계층화하려면(예: 45초에 "여전히 생각 중" 버블 보내기) 어댑터에서 _keep_typing을 override하고 super()._keep_typing()과 함께 고유한 작업을 예약한 다음 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

핵심 사항:

• 항상 await super()._keep_typing(...). 입력 하트비트는 독립적으로 유용합니다. 교체하지 말고 그 위에 레이어를 추가하세요.
• finally에서 부가 작업을 해제합니다. LLM이 완료되면(또는 /stop이 실행을 취소하면) 게이트웨이는 입력 작업을 취소합니다. 부차 작업도 취소를 관찰해야 합니다. 그렇지 않으면 응답이 이미 전달된 후에 계속 실행되고 실행될 수 있습니다.
• **사용자가 /stop을 발행할 때 고아 UX 상태를 해결하려면 interrupt_session_activity**과 페어링하세요. LINE의 경우 이는 포스트백 캐시 항목을 PENDING에서 ERROR으로 전환하여 영구 "답변 얻기" 버튼이 반복 대신 "실행이 중단되었습니다" 메시지를 전달하는 것을 의미합니다.

패턴: 즉시 보내는 대신 캐시를 통해 라우팅하는 서브클래스 send

느린 응답 UX가 나중에 검색하기 위해 응답을 캐시하는 경우(LINE의 포스트백 흐름), send 재정의는 세 가지 모드를 인식해야 합니다.

1. 이 채팅에 대한 포스트백 활성 대기 중 → request_id 아래에 응답을 캐시하고 표시되는 항목을 보내지 마세요.
2. 시스템 사용 중 응답(⚡ Interrupting, ⏳ Queued, ⏩ Steered) → 캐시를 우회하고 시각적으로 전송하여 사용자가 입력에 대한 게이트웨이의 응답을 볼 수 있도록 합니다.
3. 정상적인 응답 → 평소처럼 응답 토큰 또는 푸시를 통해 보냅니다.

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`은 게이트웨이 자체의 통화중 확인 접두사입니다(`⚡`, `⏳`, `⏩`, `💾`). 캐시된 UX 상태에 관계없이 항상 시각적으로 확인하세요.

### 이 패턴이 적합한 경우 \{#when-this-pattern-is-appropriate}

다음과 같은 경우 타이핑 루프 재정의 접근 방식을 사용하세요.

- 플랫폼의 아웃바운드 API에는 엄격한 시간 제한(일회용 응답 토큰, 고정 세션 만료 등)이 있습니다. AND
- *비행 중에 보이는 버블*은 해당 플랫폼에서 허용되는 UX입니다.

다음과 같은 경우에는 더 간단한 `slow_response_threshold = 0` 항상 푸시 경로를 사용하세요.

- 플랫폼에 의미 있는 무료와 유료 구분이 없습니다. 또는
- 사용자 커뮤니티는 대화형 중간 버블보다 "로드 중... 로드 중... 완료" 무음 응답을 선호합니다.

LINE은 두 가지를 모두 지원합니다. 무료 포스트백 가져오기의 경우 임계값 기본값은 45초이고 `LINE_SLOW_RESPONSE_THRESHOLD=0`은 '항상 푸시 폴백'으로 되돌아갑니다.

### 참조 구현 \{#reference-implementation}

전체 LINE 포스트백 구현에 대해서는 `plugins/platforms/line/adapter.py`을 참조하세요. — `RequestCache` 상태 머신(`PENDING → READY → DELIVERED` 및 `/stop`에 대한 `ERROR`), `_keep_typing` override 임계값에서 템플릿 버튼 버블을 실행하고 캐시를 통해 라우팅하는 `send` override와 고아 PENDING 항목을 해결하는 `interrupt_session_activity` 재정의를 실행합니다.

### 참조 구현(플러그인 경로) \{#reference-implementations-plugin-path}

완전한 작업 예제(외부 종속성이 없는 전체 비동기 IRC 어댑터)는 저장소의 `plugins/platforms/irc/`을 참조하세요. `plugins/platforms/teams/`은 Bot Framework/Adaptive Card를 다루고, `plugins/platforms/google_chat/`은 OAuth 기반 REST API를 다루고, `plugins/platforms/line/`은 플랫폼별 느린 LLM UX가 포함된 웹후크 기반 메시징 API를 다룹니다.

---

## 단계별 체크리스트(내장 경로) \{#step-by-step-checklist-built-in-path}

:::note
이 체크리스트는 Hermes 핵심 코드베이스에 플랫폼을 직접 추가하기 위한 것입니다. 일반적으로 공식적으로 지원되는 플랫폼의 핵심 기여자가 수행합니다. 커뮤니티/타사 플랫폼은 위의 [플러그인 경로](#plugin-path-recommended)를 사용해야 합니다.

:::
### 1. 플랫폼 열거형 \{#1-platform-enum}

`gateway/config.py`의 `Platform` 열거형에 플랫폼을 추가합니다.

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

2. 어댑터 파일

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"}

인바운드 메시지의 경우 MessageEvent을 빌드하고 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.py)

세 가지 터치포인트:

1. get_connected_platforms() — 플랫폼의 필수 자격 증명 확인을 추가하세요.
2. load_gateway_config() — 토큰 환경 맵 항목 추가: Platform.NEWPLAT: "NEWPLAT_TOKEN"
3. _apply_env_overrides() — 모든 NEWPLAT_* 환경 변수를 구성에 매핑합니다.

4. 게이트웨이 러너(gateway/run.py)

5가지 터치포인트:

1. _create_adapter() — elif platform == Platform.NEWPLAT: 브랜치 추가
2. _is_user_authorized() allowed_users 맵 — Platform.NEWPLAT: "NEWPLAT_ALLOWED_USERS"
3. _is_user_authorized()allow_all 맵 — Platform.NEWPLAT: "NEWPLAT_ALLOW_ALL_USERS"
4. 초기 환경 확인 _any_allowlist 튜플 — "NEWPLAT_ALLOWED_USERS" 추가
5. 초기 환경 확인 _allow_all 튜플 — "NEWPLAT_ALLOW_ALL_USERS" 추가
6. _UPDATE_ALLOWED_PLATFORMS Frozenset — Platform.NEWPLAT 추가

5. 크로스 플랫폼 전달

1. gateway/platforms/webhook.py — 배송 유형 튜플에 "newplat"을 추가합니다.
2. cron/scheduler.py — _KNOWN_DELIVERY_PLATFORMS Frozenset 및 _deliver_result() 플랫폼 맵에 추가

6. CLI 통합

1. hermes_cli/config.py — 모든 NEWPLAT_* 변수를 _EXTRA_ENV_KEYS에 추가합니다.
2. hermes_cli/gateway.py — 키, 라벨, 이모티콘, token_var, setup_instructions 및 vars를 사용하여 _PLATFORMS 목록에 항목을 추가합니다.
3. hermes_cli/platforms.py — 레이블 및 default_toolset이 있는 PlatformInfo 항목을 추가합니다(skills_config 및 tools_config TUI에서 사용됨).
4. hermes_cli/setup.py — _setup_newplat() 함수를 추가하고(gateway.py에 위임 가능) 메시징 플랫폼 목록에 튜플을 추가합니다.
5. hermes_cli/status.py — 플랫폼 감지 항목 추가: "NewPlat": ("NEWPLAT_TOKEN", "NEWPLAT_HOME_CHANNEL")
6. hermes_cli/dump.py — 플랫폼 감지 dict에 "newplat": "NEWPLAT_TOKEN" 추가

7. 도구

1. tools/send_message_tool.py — 플랫폼 맵에 "newplat": Platform.NEWPLAT 추가
2. tools/cronjob_tools.py — 배송 대상 설명 문자열에 newplat을 추가합니다.

8. 도구 세트

1. toolsets.py — _HERMES_CORE_TOOLS를 사용하여 "hermes-newplat" 도구 세트 정의를 추가합니다.
2. toolsets.py — "hermes-gateway" 포함 목록에 "hermes-newplat"을 추가합니다.

9. 선택 사항: 플랫폼 힌트

agent/prompt_builder.py — 플랫폼에 특정 렌더링 제한(마크다운 없음, 메시지 길이 제한 등)이 있는 경우 _PLATFORM_HINTS 사전에 항목을 추가하세요. 이는 플랫폼별 지침을 시스템 프롬프트에 삽입합니다.

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

모든 플랫폼에 힌트가 필요한 것은 아닙니다. 에이전트의 동작이 달라야 하는 경우에만 힌트를 추가하세요.

10. 테스트

다음을 포함하는 tests/gateway/test_newplat.py 생성:

• 구성에서 어댑터 구성
• 메시지 이벤트 구축
• 전송 방법(외부 API 모의)
• 플랫폼별 기능(암호화, 라우팅 등)

11. 문서

파일	추가할 내용
website/docs/user-guide/messaging/newplat.md	전체 플랫폼 설정 페이지
website/docs/user-guide/messaging/index.md	플랫폼 비교 표, 아키텍처 다이어그램, 도구 세트 표, 보안 섹션, 다음 단계 링크
website/docs/reference/environment-variables.md	모든 NEWPLAT_* 환경 변수
website/docs/reference/toolsets-reference.md	hermes-newplat 툴셋
website/docs/integrations/index.md	플랫폼 링크
website/sidebars.ts	문서 페이지의 사이드바 항목
website/docs/developer-guide/architecture.md	어댑터 수 + 목록
website/docs/developer-guide/gateway-internals.md	어댑터 파일 목록

패리티 감사

새 플랫폼 PR을 완료로 표시하기 전에 기존 플랫폼에 대해 패리티 감사를 실행하세요.

# 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
``.md` 및 `.ts` 파일에 대해 반복합니다. 각 차이를 조사하세요. 플랫폼 열거(업데이트 필요)입니까, 아니면 플랫폼별 참조(건너뛰기)입니까?

## 일반적인 패턴 \{#when-this-pattern-is-appropriate}

### 롱 폴 어댑터 \{#reference-implementation}

어댑터가 Telegram 또는 Weixin과 같은 긴 폴링을 사용하는 경우 폴링 루프 작업을 사용하세요.

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

콜백/웹훅 어댑터

플랫폼이 메시지를 엔드포인트(예: WeCom 콜백)에 푸시하는 경우 HTTP 서버를 실행합니다.

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

응답 기한이 촉박한 플랫폼(예: WeCom의 5초 제한)의 경우 항상 즉시 확인하고 나중에 API를 통해 에이전트의 응답을 사전에 전달하세요. 에이전트 세션은 3~30분 동안 실행됩니다. 콜백 응답 창 내 인라인 응답은 불가능합니다.

토큰 잠금

어댑터가 고유한 자격 증명을 사용하여 지속적인 연결을 유지하는 경우 두 프로필이 동일한 자격 증명을 사용하지 못하도록 범위 잠금을 추가하세요.

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)

참조 구현

어댑터	패턴	복잡성	다음에 대한 좋은 참고 자료
bluebubbles.py	REST + 웹훅	중간	간단한 REST API 통합
weixin.py	롱 폴링 + CDN	높음	미디어 처리, 암호화
wecom_callback.py	콜백/웹훅	중간	HTTP 서버, AES 암호화, 다중 앱
telegram.py	롱 폴링 + 봇 API	높음	그룹, 스레드가 포함된 모든 기능을 갖춘 어댑터