Telegram Botとの統合プラグイン開発

プロジェクトの背景

LLMサービスを一般的なリアルタイムチャットプラットフォーム（IM）と連携させることは、常に重要な取り組みとして注目されています。Difyプラグインのエコシステムは、よりシンプルで使いやすい連携方法を提供することを目指しています。この記事では、Telegramを例に、Telegram Botに接続するプラグインを開発する方法を詳しく解説します。

Telegramは、豊富なAPIを提供する無料のオープンなリアルタイムコミュニプラットフォームです。イベントベースの仕組みである使いやすいWebhook機能も備えています。この仕組みを利用して、次の図に示すようにTelegram Botプラグインを作成します。

連携の流れ:

ユーザーがTelegram Botを使用
ユーザーがTelegramでメッセージを送信すると、TelegramはDifyプラグインにHTTPリクエストを送信します。
メッセージがTelegram Botプラグインに転送される
メールシステムで受信者のアドレスが必要なように、Telegram Botを使用する場合、メッセージは処理のためにDifyアプリに転送される必要があります。これは、Telegram APIを通じてTelegram Webhookアドレスを設定し、そのアドレスをプラグインに入力することで実現できます。
プラグインがメッセージを受信し、Difyアプリに返す
プラグインはTelegramからのリクエストを処理し、ユーザーが入力した内容を解析します。そして、Difyアプリを呼び出して、返信内容を取得します。
Difyアプリが応答し、Telegram Botにメッセージを返す
Difyアプリからの応答を受信した後、プラグインは同じ経路でTelegram Botにメッセージを返します。これにより、ユーザーはTelegramを使いながらDifyアプリと直接対話できます。

前提条件

Telegram Botの作成
Difyプラグインのスキャフォールディングツール
Python環境（バージョン3.10以上）

Telegram Botの作成

@BotFatherのガイドに従って、新しいBotを作成してください。詳しい作成手順については、Telegramの公式ドキュメントを参照してください。

作成後、次のステップで使用するHTTP APIトークンが発行されます。

Difyプラグインのスキャフォールディングツールのインストール

詳細については、https://github.com/langgenius/dify-docs/blob/main/jp/plugins/initialize-development-tools.mdを参照してください。

Python環境の準備

詳しい手順については、Pythonインストールチュートリアルを参照するか、LLMに詳しいインストール手順を問い合わせてください。

プラグインの開発

それでは、実際のプラグインのコーディング作業を開始しましょう。始める前に、クイックスタート：拡張タイププラグインの開発を確認するか、Difyプラグインの開発経験があることを確認してください。

プロジェクトの初期化

次のコマンドを実行して、プラグイン開発プロジェクトを初期化します：

./dify-plugin-darwin-arm64 plugin init

プロンプトに従って、プロジェクトの基本情報を入力し、extensionテンプレートを選択します。権限設定では、AppsとEndpointsの両方にチェックを入れてください。

Difyプラットフォームの機能に対するプラグインからのリクエスト（逆呼び出し）については、逆呼び出し：Appを参照してください。

1. 設定フォームの編集

このプラグインでは、返信に使用するDifyアプリを指定する必要があります。また、返信時にTelegramボットトークンが必要になるため、これらの2つのフィールドをプラグインフォームに追加する必要があります。

group/your-project.yamlのように、グループパスにあるyamlファイルを編集してください。フォームの設定ファイル名は、プラグイン作成時に入力した基本情報によって決まります。対応するyamlファイルを編集してください。

コード例:

your-project.yaml

settings:
  - name: bot_token
    type: secret-input
    required: true
    label:
      en_US: Bot Token
      pt_BR: Token do Bot
      ja_JP: ボットトークン
  - name: app
    type: app-selector
    scope: chat
    required: true
    label:
      en_US: App
      ja_JP: アプリ
    placeholder:
      en_US: the app you want to use to answer telegram messages
      pt_BR: o app que você deseja usar para responder mensagens do Telegram
      ja_JP: あなたが Telegram メッセージに回答するために使用するアプリ
endpoints:
  - endpoints/your_project.yaml

コードデータ構造の説明：

  - name: app
    type: app-selector
    scope: chat

プラグインを利用すると、特定のDifyアプリにアクセスし、メッセージを転送できます。
scopeフィールドはチャットフィールドとして定義されており、エージェント、チャットボット、チャットフローといったアプリでのみ利用可能です。

例:

endpoints/your_path.yaml

path: "/your_project/message"
method: "POST"
extra:
  python:
    source: "endpoints/your_project.py"

2. 機能コードを編集する

endpoints/your_project.pyファイルを編集し、以下のコードを追加してください：

import json
import traceback
import requests
from typing import Mapping
from werkzeug import Request, Response
from dify_plugin import Endpoint

class TelegramWebhook(Endpoint):
    def _invoke(self, r: Request, values: Mapping, settings: Mapping) -> Response:
        """
        Invokes the endpoint with the given request.
        """
        data = r.get_json()

        message = data.get("message", {})
        chat = message.get("chat", {})
        chat_id = chat.get("id")
        if not chat or not message:
            return Response(status=200, response="ok")

        message_id = message.get("message_id")
        bot_token = settings.get("bot_token", "")
        chat_type = chat.get("type")

        reply_message = {
            "method": "sendMessage",
            "chat_id": chat_id,
            "reply_to_message_id": message_id,
            "text": message.get("text"),
        }
        
        return Response(
            status=200,
            response=json.dumps(reply_message),
            content_type="application/json",
        )

2.1 プラグインのデバッグ

Difyではリモートデバッグ機能が利用できます。「プラグイン管理」ページで、デバッグキーとリモートサーバーアドレスを取得してください。

プラグインプロジェクトに戻り、.env.exampleファイルをコピーして.envにリネームします。そして、リモートサーバーアドレスとデバッグキーを.envファイルに記入してください。

.envファイルの内容は以下の通りです：

INSTALL_METHOD=remote
REMOTE_INSTALL_HOST=localhost
REMOTE_INSTALL_PORT=5003
REMOTE_INSTALL_KEY=****-****-****-****-****

python -m mainコマンドを実行してプラグインを起動します。プラグインページで、プラグインがワークスペースにインストールされたことを確認できます。他のチームメンバーもこのプラグインにアクセス可能です。

python -m main

プラグインのエンドポイント設定

Difyのプラグイン管理ページで、自動的にインストールされたテストプラグインを見つけます。新しいエンドポイントを作成し、名前、Botトークンを入力し、接続先のアプリを選択してください。

プラグインからURLをコピーし、以下のコマンドと組み合わせてテストリクエストコマンドを作成します。

ここで、<hook_url>には先ほどコピーしたアドレスを、<bot_token>には準備で取得したTelegram HTTP APIトークンを入力します。

curl -F "url=<hook_url>" https://api.telegram.org/bot<bot_token>/setWebhook

実行すると、以下の結果が表示されるはずです。

Telegramボットにメッセージを送信すると、送信したメッセージがそのまま繰り返されることを確認できます。これは、プラグインがTelegramボットとの接続を正しく確立したことを示しています。

2.2 Difyアプリで返信する

Difyアプリを使用して返信するコードを追加するために、関数コードを修正します：

import json
import traceback
from typing import Mapping
from werkzeug import Request, Response
from dify_plugin import Endpoint

class TelegramWebhook(Endpoint):
    def _invoke(self, r: Request, values: Mapping, settings: Mapping) -> Response:
        """
        Invokes the endpoint with the given request.
        """
        data = r.get_json()

        message = data.get("message", {})
        chat = message.get("chat", {})
        chat_id = chat.get("id")
        if not chat or not message:
            return Response(status=200, response="ok")

        message_id = message.get("message_id")
        bot_token = settings.get("bot_token", "")
        chat_type = chat.get("type")

        try:
            response = self.session.app.chat.invoke(
                app_id=settings["app"]["app_id"],
                query=message.get("text", ""),
                inputs={},
                response_mode="blocking",
            )
            return Response(
                status=200,
                response=json.dumps({
                    "method": "sendMessage",
                    "chat_id": chat_id,
                    "reply_to_message_id": message_id,
                    "text": response.get("answer", ""),
                }),
                content_type="text/plain",
            )
        except Exception as e:
            err = traceback.format_exc()
            return Response(
                status=200,
                response="Sorry, I'm having trouble processing your request. Please try again later." + str(err),
                content_type="text/plain",
            )

このコードでは、self.session.app.chat.invoke を使用してDifyプラットフォーム内のアプリを呼び出します。app_id や query などの情報を渡し、最終的にその応答をTelegram Botに返します。

プラグインを再起動し、再度デバッグを実行すると、Telegram BotがDifyアプリからの返信メッセージを正しく出力していることを確認できます。

プラグインのパッケージ化

プラグインが正常に動作することを確認したら、以下のコマンドラインツールでプラグインをパッケージ化できます。実行すると、現在のフォルダに telegram.difypkg ファイルが作成されます。これがプラグインの最終的なパッケージです。

dify plugin package ./telegram

おめでとうございます！ツールタイプのプラグイン開発、デバッグ、パッケージ化の全工程が完了しました。

プラグインの公開

作成したプラグインは、Dify Plugins コードリポジトリにアップロードして公開できます。アップロード前に、プラグインがプラグイン公開ガイドに準拠していることを確認してください。承認されると、コードはmasterブランチにマージされ、Dify Marketplace で自動的に公開されます。

参考情報

Difyプラグインのプロジェクトコード全体を確認したい場合は、Githubコードリポジトリを参照してください。他のプラグインの完全なコードとその詳細も確認できます。

プラグインに関する詳細は、以下のコンテンツをご覧ください。

クイックスタート:

プラグイン仕様定義ドキュメント:

Previousベストプラクティス NextSlack Bot プラグインの開発

Last updated 1 day ago

settings: - name: bot_token type: secret-input required: true label: en_US: Bot Token pt_BR: Token do Bot ja_JP: ボットトークン - name: app type: app-selector scope: chat required: true label: en_US: App ja_JP: アプリ placeholder: en_US: the app you want to use to answer telegram messages pt_BR: o app que você deseja usar para responder mensagens do Telegram ja_JP: あなたが Telegram メッセージに回答するために使用するアプリ endpoints: - endpoints/your_project.yaml

import json import traceback import requests from typing import Mapping from werkzeug import Request, Response from dify_plugin import Endpoint class TelegramWebhook(Endpoint): def _invoke(self, r: Request, values: Mapping, settings: Mapping) -> Response: """ Invokes the endpoint with the given request. """ data = r.get_json() message = data.get("message", {}) chat = message.get("chat", {}) chat_id = chat.get("id") if not chat or not message: return Response(status=200, response="ok") message_id = message.get("message_id") bot_token = settings.get("bot_token", "") chat_type = chat.get("type") reply_message = { "method": "sendMessage", "chat_id": chat_id, "reply_to_message_id": message_id, "text": message.get("text"), } return Response( status=200, response=json.dumps(reply_message), content_type="application/json", )

import json import traceback from typing import Mapping from werkzeug import Request, Response from dify_plugin import Endpoint class TelegramWebhook(Endpoint): def _invoke(self, r: Request, values: Mapping, settings: Mapping) -> Response: """ Invokes the endpoint with the given request. """ data = r.get_json() message = data.get("message", {}) chat = message.get("chat", {}) chat_id = chat.get("id") if not chat or not message: return Response(status=200, response="ok") message_id = message.get("message_id") bot_token = settings.get("bot_token", "") chat_type = chat.get("type") try: response = self.session.app.chat.invoke( app_id=settings["app"]["app_id"], query=message.get("text", ""), inputs={}, response_mode="blocking", ) return Response( status=200, response=json.dumps({ "method": "sendMessage", "chat_id": chat_id, "reply_to_message_id": message_id, "text": response.get("answer", ""), }), content_type="text/plain", ) except Exception as e: err = traceback.format_exc() return Response( status=200, response="Sorry, I'm having trouble processing your request. Please try again later." + str(err), content_type="text/plain", )