channel.py (1700B)
1 import inspect 2 from sanic import Blueprint, response 3 from sanic.request import Request 4 from sanic.response import HTTPResponse 5 from typing import Text, Callable, Awaitable 6 7 from rasa.core.channels.channel import ( 8 InputChannel, 9 CollectingOutputChannel, 10 UserMessage, 11 ) 12 13 14 class Matrix(InputChannel): 15 def name(self) -> Text: 16 """Name of your custom channel.""" 17 return "matrix" 18 19 def blueprint( 20 self, on_new_message: Callable[[UserMessage], Awaitable[None]] 21 ) -> Blueprint: 22 custom_webhook = Blueprint( 23 "custom_webhook_{}".format(type(self).__name__), 24 inspect.getmodule(self).__name__, 25 ) 26 27 @custom_webhook.route("/", methods=["GET"]) 28 async def health(request: Request) -> HTTPResponse: 29 return response.json({"status": "ok"}) 30 31 @custom_webhook.route("/webhook", methods=["POST"]) 32 async def receive(request: Request) -> HTTPResponse: 33 sender_id = request.json.get("sender") # method to get sender_id 34 text = request.json.get("text") # method to fetch text 35 input_channel = self.name() # method to fetch input channel 36 metadata = self.get_metadata(request) # method to get metadata 37 38 collector = CollectingOutputChannel() 39 40 # include exception handling 41 42 await on_new_message( 43 UserMessage( 44 text, 45 collector, 46 sender_id, 47 input_channel=input_channel, 48 metadata=metadata, 49 ) 50 ) 51 52 return response.json(collector.messages) 53 54 return custom_webhook