플랫폼 어댑터 추가

기준일: 2026-08-02
난이도: 중급
공식 기준: Adding a Platform Adapter

개요

Hermes Messaging Gateway에 새 플랫폼 어댑터를 추가하는 방법입니다. 권장 경로는 플러그인(~/.hermes/plugins/<platform>/)이며, 코어 빌트인 경로(plugins/platforms/에 직접 구현)는 20개 이상의 파일을 건드려야 해 유지보수 비용이 큽니다.

핵심 개념

개념	설명
Plugin adapter	코어 수정 없이 플랫폼 추가
Built-in adapter	enum/config/runner 배선 필요
YAML→env bridge	설정 값을 env 계약으로 투영
Cron delivery	스케줄 결과 플랫폼 전달
Slow-LLM UX	typing/keepalive 패턴

구현 가이드

아키텍처 개요

모든 플랫폼 어댑터는 BasePlatformAdapter를 상속하며 다음을 구현해야 합니다.

• connect() — 연결 수립
• disconnect() — 정상 종료
• send() — 텍스트 메시지 전송
• send_typing() — 타이핑 표시 (선택)
• get_chat_info() — 메타데이터 반환 (선택)

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

플러그인 경로 (권장)

플러그인은 ~/.hermes/plugins/my-platform/ 아래 plugin.yaml(매니페스트)과 adapter.py(어댑터 클래스 + register() 진입점) 두 파일로 구성됩니다.

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

plugin.yaml은 name, label, kind: platform, version, description, author 같은 기본 메타데이터와, 플랫폼 동작에 필수인 requires_env, 부가 기능을 위한 optional_env를 선언합니다. 각 env var 항목은 단순 문자열로 쓰거나, description·prompt·url·password(민감 값 여부)를 담은 딕셔너리로 좀 더 풍부하게 쓸 수 있습니다.

adapter.py는 BasePlatformAdapter를 상속하고 PlatformConfig를 받는 어댑터 클래스, connect(is_reconnect=False)/disconnect()/send(chat_id, content, reply_to, metadata)/get_chat_info(chat_id) 비동기 메서드, 그리고 모듈 레벨 함수들(check_requirements(), validate_config(), 필요하면 _env_enablement())과 register(ctx) 진입점을 정의합니다. 다음은 그 형태를 보여주는 예시입니다.

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 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,
        required_env=["MY_PLATFORM_TOKEN"],
        install_hint="pip install my-platform-sdk",
        env_enablement_fn=_env_enablement,
        cron_deliver_env_var="MY_PLATFORM_HOME_CHANNEL",
        allowed_users_env="MY_PLATFORM_ALLOWED_USERS",
        allow_all_env="MY_PLATFORM_ALLOW_ALL_USERS",
        max_message_length=4000,
        platform_hint="You are chatting via My Platform. It supports markdown formatting.",
        emoji="💬",
    )

Configuration — 사용자는 config.yaml로 플랫폼을 설정합니다.

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

값은 어댑터 초기화 시 읽어들이는 환경변수로 대신 설정할 수도 있습니다.

플러그인 시스템이 자동으로 처리하는 것 — 레지스트리는 코어 수정 없이 다음 연동 지점을 대신 처리합니다.

연동 지점	동작 방식
게이트웨이 어댑터 생성	코어 if/elif 체인보다 먼저 레지스트리를 확인
설정 파싱	Platform._missing_()가 임의의 플랫폼 이름을 허용
연결된 플랫폼 검증	레지스트리의 validate_config() 호출
사용자 인가	allowed_users_env / allow_all_env 확인
env 전용 자동 활성화	env_enablement_fn이 PlatformConfig.extra + home_channel을 채움
YAML 설정 브리지	apply_yaml_config_fn이 config.yaml 키를 env var·extras로 변환
Cron 전달	cron_deliver_env_var가 deliver=<name>을 동작시킴
hermes config UI 항목	plugin.yaml의 requires_env / optional_env가 자동 채움
send 엔진 (tools/send_message_tool.py)	살아있는 게이트웨이 어댑터를 거쳐 라우팅
Webhook 크로스플랫폼 전달	알려진 플랫폼인지 레지스트리 확인
/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() 사용
고아 설정 경고	플러그인이 없을 때 설명 로그 출력

환경변수 기반 자동 설정

env_enablement_fn 훅은 사용자가 설정 파일을 직접 편집하지 않아도 환경변수만으로 플랫폼을 부트스트랩하게 해줍니다. 이 함수는 load_gateway_config() 안에서 어댑터가 생성되기 전에 실행되어, SDK 초기화 없이도 상태 확인과 cron 전달이 동작하게 합니다. 최소 설정이 없으면 None을 반환해 자동 활성화를 건너뛰고, 값이 있으면 PlatformConfig.extra를 채울 딕셔너리를 반환합니다. 특수 키 home_channel은 따로 추출되어 HomeChannel dataclass로 변환되고, 나머지 키는 extras에 병합됩니다.

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

YAML→env 설정 브리지

apply_yaml_config_fn 훅은 config.yaml의 플랫폼별 키를 환경변수로 번역하는 일을 플러그인이 직접 갖도록 해, 코어 설정 코드가 플랫폼마다의 스키마를 알 필요가 없게 합니다. 이 함수는 파싱된 전체 config 딕셔너리와 플랫폼의 하위 딕셔너리를 받아 os.environ을 직접 수정할 수 있고(env가 YAML보다 우선한다는 원칙은 유지), 선택적으로 PlatformConfig.extra에 병합할 딕셔너리를 반환할 수 있습니다.

def _apply_yaml_config(yaml_cfg: dict, platform_cfg: dict) -> dict | None:
    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

이 훅은 공통 키(unauthorized_dm_behavior, notice_delivery 등)에 대한 일반 처리 뒤, env var 오버라이드보다는 앞서 실행되므로 플러그인은 플랫폼 고유 키만 처리하면 됩니다. 예외는 debug 레벨로 로깅될 뿐 설정 로딩 자체를 중단시키지 않습니다.

Cron 전달

cron_deliver_env_var를 지정하면 해당 플랫폼이 유효한 cron 전달 대상으로 등록되고, 지정한 환경변수로부터 스케줄러가 home channel을 resolve할 수 있게 됩니다.

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

Out-of-process cron 전달 — cron 작업이 게이트웨이 프로세스와 분리되어 실행될 때는, standalone_sender_fn 훅으로 살아있는 어댑터 인스턴스 없이도 메시지를 보낼 수 있습니다. 빌트인 플랫폼은 tools/send_message_tool.py에 REST 헬퍼를 갖고 있지만, 플러그인은 원래 이 기능이 없었습니다.

async def _standalone_send(
    pconfig,
    chat_id,
    message,
    *,
    thread_id=None,
    media_files=None,
    force_document=False,
):
    return {"success": True, "message_id": "..."}
    # or {"error": "..."}

성공은 success: True와 message_id를 담은 딕셔너리로, 실패는 {"error": "reason"}으로 표현합니다. 예외는 잡혀서 플러그인 send 실패로 보고됩니다.

hermes config에 env 노출

CLI는 import 시점에 plugins/platforms/*/plugin.yaml을 스캔해 requires_env / optional_env 블록으로 설정 UI를 자동 채웁니다. 풍부한 딕셔너리 형식을 쓰면 설명·프롬프트·비밀번호 표시·참고 URL까지 더 나은 사용자 경험을 제공합니다.

지원되는 딕셔너리 키:

• name (필수) — 환경변수 식별자
• description — 변수 용도 설명
• prompt — 설정 화면에 보일 문구
• url — 값을 얻을 수 있는 참고 링크
• password (bool) — 민감 값 표시. 생략하면 _TOKEN/_SECRET/_KEY/_PASSWORD/_JSON 같은 접미사로 자동 판별
• category (기본값 messaging) — 분류용 그룹

문자열만 있는 항목도 동작하며, 플러그인의 label에서 설명이 자동 생성됩니다. 하위 호환을 위해 코어의 OPTIONAL_ENV_VARS 딕셔너리에 하드코딩된 항목이 우선합니다.

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

플랫폼별 느린 모델 UX

일부 플랫폼은 느린 LLM 응답을 보여주는 방식을 제약합니다. 예를 들어 LINE은 인바운드 이벤트 후 약 60초 만에 만료되는 1회용 reply token을 발급하고, WhatsApp은 24시간이 지나면 세션을 비활성으로 표시해 이후에는 템플릿 메시지만 허용하며, SMS에는 타이핑 표시나 진행 중 업데이트 개념 자체가 없습니다.

패턴 1: _keep_typing을 서브클래싱해 진행 중 UX를 얹기 — 특정 임계값(예: 45초 시점에 "아직 생각 중" 버블 전송)에서 플랫폼 고유 동작을 넣으려면 어댑터에서 _keep_typing을 오버라이드합니다. 항상 await super()._keep_typing(...)을 호출하고, 사이드 태스크는 finally에서 정리하며, 남은 UX 상태를 정리하려면 interrupt_session_activity와 짝지어 씁니다.

패턴 2: send를 서브클래싱해 즉시 전송 대신 캐시로 라우팅 — 이 채팅에 대기 중인 postback이 있으면 응답을 캐시하고, 시스템의 긴급 알림(busy-ack)이면 캐시를 우회해 바로 보이게 보내고, 평범한 응답이면 평소처럼 reply-token-or-push로 보냅니다.

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)

이 패턴이 적합한 경우 — reply token 만료나 세션 타임아웃처럼 시간에 민감한 제약이 있는 플랫폼. 참고 구현 — plugins/platforms/line/adapter.py에 LINE의 postback 처리 전체 구현이 있습니다.

빌트인 경로 단계 체크리스트

빌트인 플랫폼을 구현하려면 다음 11개 영역을 갱신해야 합니다.

1. Platform Enum — gateway/config.py에 새 플랫폼을 enum 값으로 추가합니다.

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

2. Adapter File — plugins/platforms/newplat/adapter.py에 어댑터를 구현합니다. BasePlatformAdapter를 상속하고, connect()에서 self._mark_connected()를, disconnect()에서 self._mark_disconnected()를 호출하며, 인바운드 메시지는 MessageEvent를 만들어 self.handle_message(event)로 넘깁니다.

3. Gateway Config (gateway/config.py) — 세 곳을 고칩니다: get_connected_platforms()가 해당 플랫폼의 필수 자격 증명을 확인하도록, load_gateway_config()의 토큰 환경변수 매핑에 Platform.NEWPLAT: "NEWPLAT_TOKEN"을 추가, _apply_env_overrides()에서 모든 NEWPLAT_* 환경변수를 처리.

4. Gateway Runner (gateway/run.py) — 여섯 지점을 고칩니다: _create_adapter()에 elif platform == Platform.NEWPLAT: 분기 추가, _is_user_authorized()의 allowed-users 맵에 Platform.NEWPLAT: "NEWPLAT_ALLOWED_USERS" 추가, allow-all 맵에도 동일 플랫폼 추가, 초기 환경변수 확인 튜플 _any_allowlist에 "NEWPLAT_ALLOWED_USERS" 포함, _allow_all 튜플에 "NEWPLAT_ALLOW_ALL_USERS" 추가, _UPDATE_ALLOWED_PLATFORMS frozenset에 Platform.NEWPLAT 추가.

5. Cross-Platform Delivery — 두 파일을 고칩니다: gateway/platforms/webhook.py의 전달 타입 튜플에 "newplat" 추가, cron/scheduler.py의 _KNOWN_DELIVERY_PLATFORMS frozenset과 _deliver_result() 플랫폼 맵에 해당 항목 추가.

6. CLI Integration — 여섯 CLI 파일을 고칩니다: hermes_cli/config.py(_EXTRA_ENV_KEYS에 모든 NEWPLAT_* 변수 추가), hermes_cli/gateway.py(_PLATFORMS 목록에 key·label·emoji·토큰 변수·설정 안내·환경변수 목록을 담은 항목 추가), hermes_cli/platforms.py(label과 기본 toolset을 담은 PlatformInfo 항목 추가 — skills·tools 설정 UI가 사용), hermes_cli/setup.py(_setup_newplat() 함수를 만들어 메시징 플랫폼 튜플에 추가), hermes_cli/status.py(("NEWPLAT_TOKEN", "NEWPLAT_HOME_CHANNEL") 튜플로 플랫폼 감지 추가), hermes_cli/dump.py(플랫폼 감지 딕셔너리에 "newplat": "NEWPLAT_TOKEN" 추가).

문서·사이드바까지 갱신해야 하는 파일은 다음과 같습니다.

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

Parity Audit

새 어댑터를 다 만든 뒤에는 기존 레퍼런스 플랫폼과 비교해 parity audit을 합니다. 레퍼런스 플랫폼(예: bluebubbles)을 언급하는 모든 Python·Markdown·TypeScript 파일을 찾고, 같은 방식으로 새 플랫폼을 언급하는 파일을 찾습니다. 레퍼런스 집합에는 있지만 새 플랫폼 집합에는 없는 파일은 놓친 부분일 가능성이 있습니다. 각 gap이 실제로 갱신이 필요한 플랫폼 열거인지, 아니면 건너뛰어도 되는 플랫폼 고유 참조인지 확인합니다.

# 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

Common Patterns

Long-Poll Adapters — Telegram, Weixin처럼 롱폴링을 쓰는 어댑터는 폴링 루프 태스크를 구현합니다.

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 — WeCom Callback처럼 메시지를 엔드포인트로 밀어주는 플랫폼은 HTTP 서버를 띄웁니다. WeCom의 5초 응답 기한처럼 응답 기한이 빡빡한 플랫폼은 즉시 ack하고, 에이전트 세션은 보통 3~30분이 걸리므로 실제 응답은 나중에 API로 비동기 전달합니다.

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

Token Locks — 고유 자격 증명으로 영속 연결을 유지하는 어댑터는 scoped lock으로 같은 자격 증명을 여러 프로필이 동시에 쓰지 못하게 막습니다.

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

대표적인 참고 구현은 다음과 같습니다.

어댑터	패턴	복잡도	참고하기 좋은 경우
bluebubbles.py	REST + webhook	중간	단순 REST API 연동
weixin.py	Long-poll + CDN	높음	미디어 처리, 암호화
wecom_callback.py	Callback/webhook	중간	HTTP 서버, AES 암호화, 멀티 앱
plugins/platforms/irc/adapter.py	Long-poll + IRC protocol	높음	scoped token lock을 갖춘 완전한 플러그인 어댑터

체크리스트

• 공식 원문 Adding a Platform Adapter과 대조했다
• 관련 코드·설정·권한을 로컬에서 확인했다
• 보안·opt-in·allowlist 정책을 지켰다
• 스모크 테스트 또는 단계 검증을 수행했다

다음 단계

• 공식 문서: Adding a Platform Adapter
• Hermes Agent 소개
• 트러블슈팅