notice

This is unreleased documentation for Rasa Documentation Main/Unreleased version.
For the latest released documentation, see the latest version (3.x).

Version: Main/Unreleased

rasa.core.channels.socketio

SocketBlueprint Objects

class SocketBlueprint(Blueprint)

__init__

def __init__(sio: AsyncServer, socketio_path: Text, *args: Any,
**kwargs: Any) -> None

Creates a :class:sanic.Blueprint for routing socketio connenctions.

Arguments:

  • sio: Instance of :class:socketio.AsyncServer class
  • socketio_path: string indicating the route to accept requests on.

register

def register(app: Sanic, options: Dict[Text, Any]) -> None

Attach the Socket.IO webserver to the given Sanic instance.

Arguments:

  • app: Instance of :class:sanic.app.Sanic class
  • options: Options to be used while registering the blueprint into the app.

SocketIOOutput Objects

class SocketIOOutput(OutputChannel)

send_text_message

async def send_text_message(recipient_id: Text, text: Text,
**kwargs: Any) -> None

Send a message through this channel.

send_image_url

async def send_image_url(recipient_id: Text, image: Text,
**kwargs: Any) -> None

Sends an image to the output.

send_text_with_buttons

async def send_text_with_buttons(recipient_id: Text, text: Text,
buttons: List[Dict[Text, Any]],
**kwargs: Any) -> None

Sends buttons to the output.

send_elements

async def send_elements(recipient_id: Text, elements: Iterable[Dict[Text,
Any]],
**kwargs: Any) -> None

Sends elements to the output.

send_custom_json

async def send_custom_json(recipient_id: Text, json_message: Dict[Text, Any],
**kwargs: Any) -> None

Sends custom json to the output.

send_attachment

async def send_attachment(recipient_id: Text, attachment: Dict[Text, Any],
**kwargs: Any) -> None

Sends an attachment to the user.

SocketIOInput Objects

class SocketIOInput(InputChannel)

A socket.io input channel.

__init__

def __init__(user_message_evt: Text = "user_uttered",
bot_message_evt: Text = "bot_uttered",
namespace: Optional[Text] = None,
session_persistence: bool = False,
socketio_path: Optional[Text] = "/socket.io",
jwt_key: Optional[Text] = None,
jwt_method: Optional[Text] = "HS256",
metadata_key: Optional[Text] = "metadata")

Creates a SocketIOInput object.

get_output_channel

def get_output_channel() -> Optional["OutputChannel"]

Creates socket.io output channel object.

blueprint

def blueprint(
on_new_message: Callable[[UserMessage], Awaitable[Any]]) -> Blueprint

Defines a Sanic blueprint.