[Feature]: LINE Messaging Platform Adapter

[kuei790802]

Problem or Use Case

Summary

Add a LINE Messaging API gateway adapter for Hermes Agent.

Motivation

LINE has over 200M monthly active users across Asia and is the dominant messaging platform in Taiwan, Japan, and Thailand, but Hermes Agent currently does not support it as a gateway platform.

I’m building a LINE bot for a family insurance brokerage business in Taiwan, so this is not just a theoretical integration.
I have a real production use case to validate it end-to-end.

Proposed Solution

Add a native LINE Messaging API adapter under gateway/platforms/line.py, following the existing gateway platform adapter pattern.

The implementation would likely:

• accept LINE webhook events
• verify the X-Line-Signature header before processing
• normalize incoming LINE events into Hermes gateway message handling
• use reply tokens for standard replies
• fall back to the Push Message API when a reply token has expired
• support basic access control via LINE_ALLOWED_USERS

Likely files involved:

• gateway/platforms/line.py
• gateway/run.py
• gateway/config.py
• .env.example
• tests/test_line_adapter.py
• website/docs/user-guide/messaging/line.md
• potentially pyproject.toml for an optional LINE SDK dependency

Potential environment variables:

• LINE_CHANNEL_SECRET
• LINE_CHANNEL_ACCESS_TOKEN
• LINE_ALLOWED_USERS

Alternatives Considered

I considered a few alternatives:

1. Using the existing Webhook adapter as a workaround

   • This might work for a one-off use case, but it would duplicate gateway logic instead of integrating cleanly with Hermes' existing platform architecture.

2. Waiting to see if someone else implements it

   • I checked the current Issues list and didn’t find an existing feature request for LINE gateway support.

3. Not adding LINE support at all

   • That would keep the gateway simpler, but it would leave out a major messaging platform in Asia.

Because Hermes already has a multi-platform gateway architecture, a native LINE adapter seems like the most maintainable approach.

Feature Type

Gateway / messaging improvement

Scope

Large (new module or significant refactor)

Contribution

• I'd like to implement this myself and submit a PR

[audachang]

Working LINE Bridge Solution (Standalone, No Core Changes)

While waiting for native LINE support, I built a standalone LINE bridge that connects LINE to Hermes Agent via the built-in OpenAI-compatible API server. It's running in production and works well.

Architecture

LINE users → LINE webhook (HTTPS)
  → nginx + Let's Encrypt
  → LINE Bridge (aiohttp on :8650)
  → Hermes API Server (:8642)
  → LLM

How It Works

The bridge is a ~200-line Python script (bridge.py) that:

1. Receives LINE webhook events and validates X-Line-Signature (HMAC-SHA256)
2. Forwards text messages to Hermes via POST /v1/chat/completions
3. Replies via LINE's Reply API, or falls back to Push API if the reply token expires (which happens when the LLM takes >1 minute)
4. Tracks sessions per LINE source (user/group/room) using X-Hermes-Session-Id

Quick Setup

1. Enable Hermes API server: hermes config set platforms.api_server.enabled true
2. Add LINE_CHANNEL_ACCESS_TOKEN, LINE_CHANNEL_SECRET, API_SERVER_KEY to ~/.hermes/.env
3. Run bridge as a systemd user service (uses Hermes's own venv — aiohttp is already included)
4. Set up nginx + Let's Encrypt for HTTPS. Tip: sslip.io gives you a free domain based on your IP
5. Set LINE webhook URL to https://your-domain/webhook

Limitations vs. Native Adapter

This bridge covers bidirectional text messaging (1:1, groups, rooms) but lacks:

• Image/sticker/file message support
• Cron job delivery to LINE (no send_message tool integration)
• LINE_ALLOWED_USERS access control
• Session reset commands (/new, /reset)

A native gateway/platforms/line.py adapter would address all of these.

Gotchas

• If Tailscale runs on your server, it may hold port 443 — free it with sudo tailscale funnel --https=443 off
• Cloud firewalls (Hetzner, AWS, GCP) must allow TCP 80+443 inbound — OS firewall alone isn't enough
• LINE reply tokens expire in ~1 min; the bridge auto-falls back to push messages

Full bridge.py source code (click to expand)

#!/usr/bin/env python3
"""LINE <-> Hermes Agent Bridge"""
import asyncio, hashlib, hmac, json, logging, os, sys, base64
from aiohttp import web, ClientSession

LINE_CHANNEL_ACCESS_TOKEN = os.getenv("LINE_CHANNEL_ACCESS_TOKEN", "")
LINE_CHANNEL_SECRET = os.getenv("LINE_CHANNEL_SECRET", "")
HERMES_API_URL = os.getenv("HERMES_API_URL", "http://127.0.0.1:8642")
HERMES_API_KEY = os.getenv("API_SERVER_KEY", "")
BRIDGE_PORT = int(os.getenv("LINE_BRIDGE_PORT", "8650"))
LINE_API = "https://api.line.me/v2/bot"
logging.basicConfig(level="INFO", format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("line-bridge")
_sessions = {}

def _source_id(event):
    s = event.get("source", {})
    t = s.get("type", "user")
    if t == "group": return f"line-group-{s.get('groupId','?')}"
    if t == "room": return f"line-room-{s.get('roomId','?')}"
    return f"line-user-{s.get('userId','?')}"

def _verify_sig(body, sig):
    if not LINE_CHANNEL_SECRET: return True
    h = hmac.new(LINE_CHANNEL_SECRET.encode(), body, hashlib.sha256).digest()
    return hmac.compare_digest(sig, base64.b64encode(h).decode())

async def _line_reply(http, token, text):
    msgs = [{"type":"text","text":text[i:i+5000]} for i in range(0,len(text),5000)][:5]
    async with http.post(f"{LINE_API}/message/reply",
        json={"replyToken":token,"messages":msgs},
        headers={"Authorization":f"Bearer {LINE_CHANNEL_ACCESS_TOKEN}","Content-Type":"application/json"}) as r:
        return r.status == 200

async def _line_push(http, to, text):
    msgs = [{"type":"text","text":text[i:i+5000]} for i in range(0,len(text),5000)][:5]
    async with http.post(f"{LINE_API}/message/push",
        json={"to":to,"messages":msgs},
        headers={"Authorization":f"Bearer {LINE_CHANNEL_ACCESS_TOKEN}","Content-Type":"application/json"}) as r:
        return r.status == 200

async def _ask_hermes(http, msg, src):
    hdrs = {"Content-Type":"application/json"}
    if HERMES_API_KEY: hdrs["Authorization"] = f"Bearer {HERMES_API_KEY}"
    if src in _sessions: hdrs["X-Hermes-Session-Id"] = _sessions[src]
    async with http.post(f"{HERMES_API_URL}/v1/chat/completions",
        json={"model":"hermes-agent","messages":[{"role":"user","content":msg}],"stream":False},
        headers=hdrs, timeout=300) as r:
        if r.status != 200: return f"Error ({r.status})"
        data = await r.json()
        sid = r.headers.get("X-Hermes-Session-Id")
        if sid: _sessions[src] = sid
        return data.get("choices",[{}])[0].get("message",{}).get("content","(empty)")

async def handle_webhook(req):
    body = await req.read()
    if not _verify_sig(body, req.headers.get("X-Line-Signature","")): return web.json_response({"error":"bad sig"},status=401)
    events = json.loads(body).get("events",[])
    if not events: return web.json_response({"status":"ok"})
    for e in events:
        if e.get("type")=="message" and e.get("message",{}).get("type")=="text":
            src = _source_id(e)
            push_to = e["source"].get("groupId") or e["source"].get("roomId") or e["source"].get("userId","")
            asyncio.create_task(_process(e["message"]["text"], e.get("replyToken",""), src, push_to))
    return web.json_response({"status":"ok"})

async def _process(text, token, src, push_to):
    async with ClientSession() as http:
        resp = await _ask_hermes(http, text, src)
        if not await _line_reply(http, token, resp) and push_to:
            await _line_push(http, push_to, resp)

def main():
    assert LINE_CHANNEL_ACCESS_TOKEN, "LINE_CHANNEL_ACCESS_TOKEN not set"
    assert LINE_CHANNEL_SECRET, "LINE_CHANNEL_SECRET not set"
    app = web.Application()
    app.router.add_post("/webhook", handle_webhook)
    app.router.add_get("/health", lambda r: web.json_response({"status":"ok"}))
    logger.info("LINE bridge on :%d -> %s", BRIDGE_PORT, HERMES_API_URL)
    web.run_app(app, host="0.0.0.0", port=BRIDGE_PORT, print=None)

if __name__ == "__main__": main()

systemd service file (click to expand)

# ~/.config/systemd/user/line-bridge.service
[Unit]
Description=LINE Bridge for Hermes Agent
After=hermes-gateway.service

[Service]
Type=simple
ExecStart=%h/.hermes/hermes-agent/venv/bin/python %h/.hermes/line-bridge/bridge.py
EnvironmentFile=%h/.hermes/.env
Restart=on-failure
RestartSec=5

[Install]
WantedBy=default.target

Happy to contribute this as a PR for docs, or help build the native adapter if maintainers are interested.

[kuei790802]

Wow, thank you so much for this @yuga-hashimoto!

I was actually waiting for an official response and planning to build a basic adapter myself, but your implementation is so much more comprehensive. It's a huge and pleasant surprise to see this done so perfectly!

I will pull your branch to my local machine and test it with my LINE bot setup this week. Will report back if I find any edge cases.

Really looking forward to seeing this merged. This will be a game-changer for building automated agents in the APAC region where LINE is heavily used!