Version: Latest

Custom Connectors

You can implement your own custom channel connector as a python class. You can use the rasa.core.channels.rest.RestInput class as a template.

A custom connector class must subclass rasa.core.channels.channel.InputChannel and implement at least a blueprint and name method.

The name method

The name method defines the url prefix for the connector's webhook. It also defines the channel name you should use in any channel specific response variations and the name you should pass to the output_channel query parameter on the trigger intent endpoint.

For example, if your custom channel is named myio, you would define the name method as:

from rasa.core.channels.channel import InputChannel
class MyIO(InputChannel):
def name() -> Text:
"""Name of your custom channel."""
return "myio"

You would write a response variation specific to the myio channel as:

domain.yml
responses:
utter_greet:
- text: Hi! I'm the default greeting.
- text: Hi! I'm the custom channel greeting
channel: myio

The webhook you give to the custom channel to call would be http://<host>:<port>/webhooks/myio/webhook, replacing the host and port with the appropriate values from your running Rasa server.

The blueprint method

The blueprint method needs to create a sanic blueprint that can be attached to a sanic server. Your blueprint should have at least the two routes: health on the route /, and receive on the route /webhook (see example custom channel below).

As part of your implementation of the receive endpoint, you will need to tell Rasa to handle the user message. You do this by calling

on_new_message(
rasa.core.channels.channel.UserMessage(
text,
output_channel,
sender_id
)
)

Calling on_new_message will send the user message to the handle_message method.

See more details on the UserMessage object here.

The output_channel argument refers to an output channel implementing the OutputChannel class. You can either implement your own output channel class with the methods for your particular chat channel (e.g. methods to send text and images) or you can use the CollectingOutputChannel to collect the bot responses Rasa creates while the bot is processing your messages and return them as part of your endpoint response. This is the way the RestInput channel is implemented. For examples on how to create and use your own output channel, take a look at the implementations of the other output channels, e.g. the SlackBot in rasa.core.channels.slack.

Here is a simplified example of a custom channel connector that makes use of the CollectingOutputChannel:

custom_channel.py
import asyncio
import inspect
from sanic import Sanic, Blueprint, response
from sanic.request import Request
from sanic.response import HTTPResponse
from typing import Text, Dict, Any, Optional, Callable, Awaitable, NoReturn
import rasa.utils.endpoints
from rasa.core.channels.channel import (
InputChannel,
CollectingOutputChannel,
UserMessage,
)
class MyIO(InputChannel):
def name() -> Text:
"""Name of your custom channel."""
return "myio"
def blueprint(
self, on_new_message: Callable[[UserMessage], Awaitable[None]]
) -> Blueprint:
custom_webhook = Blueprint(
"custom_webhook_{}".format(type(self).__name__),
inspect.getmodule(self).__name__,
)
@custom_webhook.route("/", methods=["GET"])
async def health(request: Request) -> HTTPResponse:
return response.json({"status": "ok"})
@custom_webhook.route("/webhook", methods=["POST"])
async def receive(request: Request) -> HTTPResponse:
sender_id = request.json.get("sender") # method to get sender_id
text = request.json.get("text") # method to fetch text
input_channel = self.name() # method to fetch input channel
metadata = self.get_metadata(request) # method to get metadata
collector = CollectingOutputChannel()
# include exception handling
await on_new_message(
UserMessage(
text,
collector,
sender_id,
input_channel=input_channel,
metadata=metadata,
)
)
return response.json(collector.messages)
return custom_webhook
New in 3.9

You can now access all the data about the current conversation via the tracker_state property of any OutputChannel subclass.

Data such as slots, active flows, intents, custom actions called, etc are all present in the current state of the tracker and these are now easily accessible for use.

Here is a simplified example of a custom channel connector that makes use of the CollectingOutputChannel to collect the bot response and returns the tracker_state in the API response using the CollectingOutputChannel.tracker_state field:

custom_channel.py
import inspect
from sanic import Blueprint, response
from sanic.request import Request
from sanic.response import HTTPResponse
from typing import Text, Callable, Awaitable
import rasa.utils.endpoints
from rasa.core.channels.channel import (
InputChannel,
CollectingOutputChannel,
UserMessage,
)
class MyIO(InputChannel):
def name() -> Text:
"""Name of your custom channel."""
return "myio"
def blueprint(
self, on_new_message: Callable[[UserMessage], Awaitable[None]]
) -> Blueprint:
"""Groups the collection of endpoints used by input channel."""
custom_webhook = Blueprint(
"custom_webhook_{}".format(type(self).__name__),
inspect.getmodule(self).__name__,
)
@custom_webhook.route("/", methods=["GET"])
async def health(request: Request) -> HTTPResponse:
return response.json({"status": "ok"})
@custom_webhook.route("/webhook", methods=["POST"])
async def receive(request: Request) -> HTTPResponse:
sender_id = request.json.get("sender") # method to get sender_id
text = request.json.get("text") # method to fetch text
input_channel = self.name() # method to fetch input channel
metadata = self.get_metadata(request) # method to get metadata
collector = CollectingOutputChannel()
# include exception handling
await on_new_message(
UserMessage(
text,
collector,
sender_id,
input_channel=input_channel,
metadata=metadata,
headers=request.headers,
)
)
# build the response dict with the tracker_state and other data
response_dict = {
"messages": collector.messages,
"metadata": metadata,
"conversation_id": sender_id,
"tracker_state": collector.tracker_state,
}
return response.json(response_dict)
return custom_webhook

Metadata on messages

If you need to use extra information from your front end in your custom actions, you can pass this information using the metadata key of your user message. This information will accompany the user message through the Rasa server into the action server when applicable, where you can find it stored in the tracker. Message metadata will not directly affect NLU classification or action prediction.

The InputChannel class's default implementation of get_metadata ignores all metadata. To extract metadata in a custom connector, implement the get_metadata method. The SlackInput channel provides one example of a get_metadata method that extracts metadata according to the channel's response format.

Credentials for Custom Channels

To use a custom channel, you need to supply credentials for it in a credentials configuration file called credentials.yml. This credentials file has to contain the module path (not the channel name) of your custom channel and any required configuration parameters.

For example, for a custom connector class called MyIO saved in a file addons/custom_channel.py, the module path would be addons.custom_channel.MyIO, and the credentials could look like:

credentials.yml
addons.custom_channel.MyIO:
username: "user_name"
another_parameter: "some value"

To make the Rasa server aware of your custom channel, specify the path to credentials.yml to the Rasa server at startup with the command line argument --credentials .

Testing the Custom Connector Webhook

To test your custom connector, you can POST messages to the webhook using a json body with the following format:

{
"sender": "test_user", // sender ID of the user sending the message
"message": "Hi there!",
"metadata": {} // optional, any extra info you want to add for processing in NLU or custom actions
}

For a locally running Rasa server, the curl request would look like this:

curl --request POST \
--url http://localhost:5005/webhooks/myio/webhook \
--header 'Content-Type: application/json' \
--data '{
"sender": "test_user",
"message": "Hi there!",
"metadata": {}
}'