ボット開発

Webhook

getUpdates の代わりに、サーバーから新着をボットの HTTPS アドレスへ直接送らせる方法 — 設定、検証、再試行、レスポンスでの返信。

getUpdates は、ボットがサーバーに問い合わせ続ける方式です。Webhook はその逆で、新着が発生すると Pabal サーバーがボットの HTTPS アドレスに直接送る方式です。ボットが常時稼働しているサーバー（クラウド・社内サーバー）で動くなら、Webhook のほうが適しています。

Webhook のやり取り

Webhook アドレスの条件

• https:// のアドレスである必要があります。証明書は公的な認証局（Let's Encrypt など）から発行されたものでなければならず、自己署名証明書のアップロード（certificate）はサポートしていません。
• インターネットから到達できるグローバルアドレスである必要があります。サーバーは、ループバック（127.0.0.1、::1）、プライベートネットワーク（10.、172.16–31.、192.168.）、リンクローカル（169.254.、クラウドのメタデータ）、CGNAT（100.64/10）のような内部アドレスには送りません。ドメインがそのようなアドレスを指している場合も拒否し、送信のたびに確認し直します。
• ポートに制限はありません（443 でなくてもかまいません）。アドレスにユーザー名・パスワード（https://user:pw@…）を含めることはできません。
• リダイレクト（3xx）はたどらず、失敗として扱います。最終的なアドレスを指定してください。
• 30 秒以内に応答する必要があります。時間のかかる処理は、先に 200 で応答してから、バックグラウンドで行ってください。

設定する

1. Webhook を受け取るプログラムを起動します。下の例のいずれかを、ボットのサーバーで 127.0.0.1:8081 で起動するとします。
2. 前段に HTTPS を置きます。たとえば Caddy なら、この 2 行だけで証明書の取得まで自動で行われます。

   bot.example.com {
       reverse_proxy 127.0.0.1:8081
   }

3. シークレットトークンを 1 つ作ります。英字・数字・_・- で 1〜256 文字です。サーバーはリクエストのたびにこの値をヘッダーに入れて送るので、リクエストが本物の Pabal サーバーから来たかどうかを確認できます。

   export WEBHOOK_SECRET=$(openssl rand -hex 32)

4. setWebhook を呼びます。

   curl -s "https://pabal.me/bot$BOT_TOKEN/setWebhook" -H 'Content-Type: application/json' -d "{
     \"url\": \"https://bot.example.com/pabal-webhook\",
     \"secret_token\": \"$WEBHOOK_SECRET\",
     \"allowed_updates\": [\"message\", \"callback_query\"],
     \"drop_pending_updates\": true
   }"
   # {"ok":true,"result":true}

5. 状態を確認します。アプリからボットにメッセージを送り、getWebhookInfo を確認します。

   curl -s "https://pabal.me/bot$BOT_TOKEN/getWebhookInfo"

   {
     "ok": true,
     "result": {
       "url": "https://bot.example.com/pabal-webhook",
       "has_custom_certificate": false,
       "pending_update_count": 0,
       "ip_address": "198.51.100.7",
       "max_connections": 40,
       "allowed_updates": ["message", "callback_query"]
     }
   }

   pending_update_count が 0 なら、正しく受け取れています。last_error_date・last_error_message は最後の失敗の記録なので、回復したあとも残っています — 時刻を見て判断してください。

setWebhook のパラメーター

Webhook の設定はサーバーに保存されるため、サーバーを再起動しても維持され、起動するとすぐに送信を再開します。ただし、まだ配信できていないアップデートはメモリ上にあるため、再起動時に消えます。

サーバーが送るリクエスト

POST /pabal-webhook HTTP/1.1
Host: bot.example.com
Content-Type: application/json
X-Telegram-Bot-Api-Secret-Token: 3f1c…(setWebhook で指定した値)

{"update_id":12,"message":{"message_id":3,"from":{"id":100001,"is_bot":false,"first_name":"ハナ"},"chat":{"id":100001,"first_name":"ハナ","type":"private"},"date":1789805661,"text":"やあ"}}

• 本文は、getUpdates の結果の要素 1 つとまったく同じ Update オブジェクトです。
• シークレットトークンを必ず確認してください。ヘッダーがないか値が異なる場合は、401 で拒否します。比較には、処理時間が一定の関数（hmac.compare_digest、crypto.timingSafeEqual）を使います。
• 同じアップデートが 2 回届くことがあります（レスポンスがサーバーに届く前に接続が切れた場合）。update_id で処理済みのものを読み飛ばせば安全です。

レスポンスで返信する

200 レスポンスの本文に Bot API の呼び出しを 1 つ入れると、サーバーがその呼び出しをボットの代わりに実行します。method にメソッド名を入れ、残りのパラメーターも一緒に入れます。

{"method": "sendMessage", "chat_id": 100001, "text": "こんにちは！"}

• 本文の形式は、JSON、フォーム、multipart/form-data のいずれも使えます（aiogram は multipart で送ります）。サイズは 1MB までです。
• この呼び出しの結果やエラーはボットには返りません（サーバーのログにだけ残ります）。結果が必要なら、通常どおり別途リクエストしてください。
• 返信することがなければ、本文なしで 200 だけを返せば十分です。

例

4 つの例はいずれもシークレットトークンを確認し、テキストメッセージを受け取るとレスポンスに sendMessage を入れてオウム返しします。前段に HTTPS（例：Caddy）を置き、127.0.0.1:8081 で受け取ることを前提にしています。

# webhook.py — 標準ライブラリのみ
# 実行: WEBHOOK_SECRET='…' python3 webhook.py
import hmac
import json
import os
from http.server import BaseHTTPRequestHandler, HTTPServer

SECRET = os.environ["WEBHOOK_SECRET"]


class Webhook(BaseHTTPRequestHandler):
    def do_POST(self):
        got = self.headers.get("X-Telegram-Bot-Api-Secret-Token", "")
        if not hmac.compare_digest(got, SECRET):
            self.send_response(401)
            self.end_headers()
            return
        update = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
        message = update.get("message")
        answer = b""
        if message and "text" in message:
            answer = json.dumps({"method": "sendMessage",
                                 "chat_id": message["chat"]["id"],
                                 "text": message["text"]}).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(answer)))
        self.end_headers()
        self.wfile.write(answer)


HTTPServer(("127.0.0.1", 8081), Webhook).serve_forever()

// webhook.mjs — Node.js 18 以上、ライブラリなし
// 実行: WEBHOOK_SECRET='…' node webhook.mjs
import http from 'node:http';
import { timingSafeEqual } from 'node:crypto';

const SECRET = Buffer.from(process.env.WEBHOOK_SECRET);

http.createServer(async (req, res) => {
  const got = Buffer.from(req.headers['x-telegram-bot-api-secret-token'] ?? '');
  if (req.method !== 'POST' || got.length !== SECRET.length || !timingSafeEqual(got, SECRET)) {
    res.writeHead(401).end();
    return;
  }
  let body = '';
  for await (const chunk of req) body += chunk;
  const message = JSON.parse(body).message;
  if (message?.text) {
    res.writeHead(200, { 'Content-Type': 'application/json' });
    res.end(JSON.stringify({ method: 'sendMessage', chat_id: message.chat.id, text: message.text }));
  } else {
    res.writeHead(200).end();
  }
}).listen(8081, '127.0.0.1');

# webhook_aiogram.py — pip install aiogram
# Webhook の設定（setWebhook）までこのプログラムが行います
import os

from aiogram import Bot, Dispatcher
from aiogram.client.session.aiohttp import AiohttpSession
from aiogram.client.telegram import TelegramAPIServer
from aiogram.webhook.aiohttp_server import SimpleRequestHandler, setup_application
from aiohttp import web

URL = "https://bot.example.com/pabal-webhook"
SECRET = os.environ["WEBHOOK_SECRET"]
dp = Dispatcher()


@dp.message()
async def echo(message):
    if message.text:
        return message.answer(message.text)       # return: Webhook のレスポンスに入れて送る


async def on_startup(bot: Bot):
    await bot.set_webhook(URL, secret_token=SECRET)


def main():
    session = AiohttpSession(api=TelegramAPIServer.from_base("https://pabal.me"))
    bot = Bot(os.environ["BOT_TOKEN"], session=session)
    dp.startup.register(on_startup)
    app = web.Application()
    SimpleRequestHandler(dispatcher=dp, bot=bot, secret_token=SECRET).register(app, path="/pabal-webhook")
    setup_application(app, dp, bot=bot)
    web.run_app(app, host="127.0.0.1", port=8081)


main()

# webhook_ptb.py — pip install "python-telegram-bot[webhooks]"
# Webhook の設定（setWebhook）までこのプログラムが行います
import os

from telegram import Update
from telegram.ext import Application, ContextTypes, MessageHandler, filters

SERVER = "https://pabal.me"


async def echo(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text(update.message.text)


app = (Application.builder().token(os.environ["BOT_TOKEN"])
       .base_url(f"{SERVER}/bot").base_file_url(f"{SERVER}/file/bot").build())
app.add_handler(MessageHandler(filters.TEXT, echo))
app.run_webhook(listen="127.0.0.1", port=8081, url_path="pabal-webhook",
                webhook_url="https://bot.example.com/pabal-webhook",
                secret_token=os.environ["WEBHOOK_SECRET"])

状態の確認 — getWebhookInfo

last_error_message に表示される内容

getUpdates に戻す

curl -s "https://pabal.me/bot$BOT_TOKEN/deleteWebhook"
# 待機中のアップデートも破棄するには: deleteWebhook?drop_pending_updates=true

Webhook がすでに受け取った（2xx で応答した）アップデートが、getUpdates で再び届くことはありません。まだ配信できていないものだけを、getUpdates で引き続き受け取ります。

開発中に試す

• トンネル：自分のコンピューターの 127.0.0.1:8081 に公開 HTTPS アドレスを割り当てるツール（cloudflared、ngrok など）を使えば、本番の Pabal サーバーでも試せます。
• 自分の Pabal サーバー：開発用の Pabal サーバーを自分で起動したなら、TELEGRAM_WEBHOOK_ALLOW_LOCAL=true で有効にしてください。そのサーバーは http://127.0.0.1:8081/… のようなローカルアドレスにも送ります。本番サーバーでは絶対に有効にしないでください — 1 つのボットが、サーバーの内部ネットワーク（データベース、クラウドのメタデータ）にリクエストを送れるようになってしまいます。

運用チェックリスト

• シークレットトークンを決め、すべてのリクエストで確認している。
• 30 秒以内に応答する — 時間のかかる処理はキューに回し、すぐに 200 を返す。
• update_id で重複を取り除く。
• Webhook が長く止まると後続のアップデートがたまり、サーバーの再起動時に消える — Webhook のプログラムを監視する（getWebhookInfo の pending_update_count、last_error_date）。
• トークンが漏れたら、/revoke のあと新しいトークンで setWebhook を呼び直す。シークレットトークンも一緒に変える。