Creating a new output adapter
You can write your own output adapters by creating a new class that inherits from chatterbot.output.OutputAdapter and overrides the necessary methods established in the OutputAdapter class.
To create your own output adapter you must override the process_response method to return a Statement object.
Note that you may need to extend the __init__ method of your custom output adapter if you intend to save a kwarg parameter that was passed into the chat bot’s constructor. (An API key might be an example of a parameter you would want to access here.)
from chatterbot.adapters import Adapter
class OutputAdapter(Adapter):
"""
A generic class that can be overridden by a subclass to provide extended
functionality, such as delivering a response to an API endpoint.
"""
def process_response(self, statement, session_id=None):
"""
Override this method in a subclass to implement customized functionality.
:param statement: The statement that the chat bot has produced in response to some input.
:param session_id: The unique id of the current chat session.
:returns: The response statement.
"""
return statementOutput format adapter
A generic class that can be overridden by a subclass to provide extended functionality, such as delivering a response to an API endpoint.
from chatterbot.adapters import Adapter
class OutputAdapter(Adapter):
"""
A generic class that can be overridden by a subclass to provide extended
functionality, such as delivering a response to an API endpoint.
"""
def process_response(self, statement, session_id=None):
"""
Override this method in a subclass to implement customized functionality.
:param statement: The statement that the chat bot has produced in response to some input.
:param session_id: The unique id of the current chat session.
:returns: The response statement.
"""The output adapter allows the chat bot to return a response in as a Statement object.
chatbot = ChatBot(
"My ChatterBot",
output_adapter="chatterbot.output.OutputAdapter",
output_format="text"
)Terminal output adapter
A simple adapter that allows ChatterBot to communicate through the terminal.
from __future__ import unicode_literals
from .output_adapter import OutputAdapter
class TerminalAdapter(OutputAdapter):
"""
A simple adapter that allows ChatterBot to
communicate through the terminal.
"""
def process_response(self, statement, session_id=None):
"""
Print the response to the user's input.
"""
print(statement.text)
return statement.textThe output terminal adapter allows a user to type into their terminal to communicate with the chat bot.
chatbot = ChatBot(
"My ChatterBot",
output_adapter="chatterbot.output.TerminalAdapter"
)Gitter output adapter
An output adapter that allows a ChatterBot instance to send responses to a Gitter room.
from __future__ import unicode_literals
from .output_adapter import OutputAdapter
class Gitter(OutputAdapter):
"""
An output adapter that allows a ChatterBot instance to send
responses to a Gitter room.
"""
def __init__(self, **kwargs):
super(Gitter, self).__init__(**kwargs)
self.gitter_host = kwargs.get('gitter_host', 'https://api.gitter.im/v1/')
self.gitter_room = kwargs.get('gitter_room')
self.gitter_api_token = kwargs.get('gitter_api_token')
authorization_header = 'Bearer {}'.format(self.gitter_api_token)
self.headers = {
'Authorization': authorization_header,
'Content-Type': 'application/json; charset=utf-8',
'Accept': 'application/json'
}
# Join the Gitter room
room_data = self.join_room(self.gitter_room)
self.room_id = room_data.get('id')
def _validate_status_code(self, response):
code = response.status_code
if code not in [200, 201]:
raise self.HTTPStatusException('{} status code recieved'.format(code))
def join_room(self, room_name):
"""
Join the specified Gitter room.
"""
import requests
endpoint = '{}rooms'.format(self.gitter_host)
response = requests.post(
endpoint,
headers=self.headers,
json={'uri': room_name}
)
self.logger.info('{} status joining room {}'.format(
response.status_code, endpoint
))
self._validate_status_code(response)
return response.json()
def send_message(self, text):
"""
Send a message to a Gitter room.
"""
import requests
endpoint = '{}rooms/{}/chatMessages'.format(self.gitter_host, self.room_id)
response = requests.post(
endpoint,
headers=self.headers,
json={'text': text}
)
self.logger.info('{} sending message to {}'.format(
response.status_code, endpoint
))
self._validate_status_code(response)
return response.json()
def process_response(self, statement, session_id=None):
self.send_message(statement.text)
return statement
class HTTPStatusException(Exception):
"""
Exception raised when unexpected non-success HTTP
status codes are returned in a response.
"""
def __init__(self, value):
self.value = value
def __str__(self):
return repr(self.value)chatbot = ChatBot(
"My ChatterBot",
output_adapter="chatterbot.output.Gitter",
gitter_api_token="my-gitter-api-token",
gitter_room="my-room-name",
gitter_only_respond_to_mentions=True,
)HipChat output adapter
An output adapter that allows a ChatterBot instance to send responses to a HipChat room.
from __future__ import unicode_literals
import json
from .output_adapter import OutputAdapter
class HipChat(OutputAdapter):
"""
An output adapter that allows a ChatterBot instance to send
responses to a HipChat room.
"""
def __init__(self, **kwargs):
super(HipChat, self).__init__(**kwargs)
self.hipchat_host = kwargs.get("hipchat_host")
self.hipchat_access_token = kwargs.get("hipchat_access_token")
self.hipchat_room = kwargs.get("hipchat_room")
authorization_header = "Bearer {}".format(self.hipchat_access_token)
self.headers = {
'Authorization': authorization_header,
'Content-Type': 'application/json'
}
import requests
self.session = requests.Session()
self.session.verify = kwargs.get('ssl_verify', True)
def send_message(self, room_id_or_name, message):
"""
Send a message to a HipChat room.
https://www.hipchat.com/docs/apiv2/method/send_message
"""
message_url = "{}/v2/room/{}/message".format(
self.hipchat_host,
room_id_or_name
)
response = self.session.post(
message_url,
headers=self.headers,
data=json.dumps({
'message': message
})
)
return response.json()
def reply_to_message(self):
"""
The HipChat api supports responding to a given message.
This may be a good feature to implement in the future to
help with multi-user conversations.
https://www.hipchat.com/docs/apiv2/method/reply_to_message
"""
raise self.AdapterMethodNotImplementedError()
def process_response(self, statement, session_id=None):
data = self.send_message(self.hipchat_room, statement.text)
# Update the output statement with the message id
self.chatbot.storage.update(
statement.add_extra_data('hipchat_message_id', data['id'])
)
return statementThis is an output adapter that allows a ChatterBot instance to send responses to a HipChat room.
Be sure to also see the documentation for the HipChat input adapter.
chatbot = ChatBot(
"My ChatterBot",
output_adapter="chatterbot.output.HipChat",
hipchat_host="https://mydomain.hipchat.com",
hipchat_room="my-room-name",
hipchat_access_token="my-hipchat-access-token",
)Microsoft Bot Framework output adapte
An output adapter that allows a ChatterBot instance to send responses to a Microsoft bot using Direct Line client protocol.
from __future__ import unicode_literals
import json
from .output_adapter import OutputAdapter
class Microsoft(OutputAdapter):
"""
An output adapter that allows a ChatterBot instance to send
responses to a Microsoft bot using *Direct Line client protocol*.
"""
def __init__(self, **kwargs):
super(Microsoft, self).__init__(**kwargs)
self.directline_host = kwargs.get(
'directline_host',
'https://directline.botframework.com'
)
self.direct_line_token_or_secret = kwargs.get(
'direct_line_token_or_secret'
)
self.conversation_id = kwargs.get('conversation_id')
authorization_header = 'BotConnector {}'.format(
self.direct_line_token_or_secret
)
self.headers = {
'Authorization': authorization_header,
'Content-Type': 'application/json'
}
def _validate_status_code(self, response):
status_code = response.status_code
if status_code not in [200, 204]:
raise self.HTTPStatusException('{} status code recieved'.format(status_code))
def get_most_recent_message(self):
"""
Return the most recently sent message.
"""
import requests
endpoint = '{host}/api/conversations/{id}/messages'.format(
host=self.directline_host,
id=self.conversation_id
)
response = requests.get(
endpoint,
headers=self.headers,
verify=False
)
self.logger.info('{} retrieving most recent messages {}'.format(
response.status_code, endpoint
))
self._validate_status_code(response)
data = response.json()
if data['messages']:
last_msg = int(data['watermark'])
return data['messages'][last_msg - 1]
return None
def send_message(self, conversation_id, message):
"""
Send a message to a HipChat room.
https://www.hipchat.com/docs/apiv2/method/send_message
"""
import requests
message_url = "{host}/api/conversations/{conversationId}/messages".format(
host=self.directline_host,
conversationId=conversation_id
)
response = requests.post(
message_url,
headers=self.headers,
data=json.dumps({
'message': message
})
)
self.logger.info('{} sending message {}'.format(
response.status_code, message_url
))
self._validate_status_code(response)
# Microsoft return 204 on operation succeeded and no content was returned.
return self.get_most_recent_message()
def process_response(self, statement, session_id=None):
data = self.send_message(self.conversation_id, statement.text)
self.logger.info('processing user response {}'.format(data))
return statement
class HTTPStatusException(Exception):
"""
Exception raised when unexpected non-success HTTP
status codes are returned in a response.
"""
def __init__(self, value):
self.value = value
def __str__(self):
return repr(self.value)This is an output adapter that allows a ChatterBot instance to send responses to a Microsoft using Direct Line protocol.
Be sure to also see the documentation for the Microsoft input adapter.
chatbot = ChatBot(
"My ChatterBot",
output_adapter="chatterbot.output.Microsoft",
directline_host="https://directline.botframework.com",
conversation_id="IEyJvnDULgn",
direct_line_token_or_secret="RCurR_XV9ZA.cwA.BKA.iaJrC8xpy8qbOF5xnR2vtCX7CZj0LdjAPGfiCpg4Fv0",
)Mailgun output adapter
A generic class that can be overridden by a subclass to provide extended functionality, such as delivering a response to an API endpoint.
from __future__ import unicode_literals
from .output_adapter import OutputAdapter
class Mailgun(OutputAdapter):
def __init__(self, **kwargs):
super(Mailgun, self).__init__(**kwargs)
# Use the bot's name for the name of the sender
self.name = kwargs.get('name')
self.from_address = kwargs.get('mailgun_from_address')
self.api_key = kwargs.get('mailgun_api_key')
self.endpoint = kwargs.get('mailgun_api_endpoint')
self.recipients = kwargs.get('mailgun_recipients')
def send_message(self, subject, text, from_address, recipients):
"""
* subject: Subject of the email.
* text: Text body of the email.
* from_email: The email address that the message will be sent from.
* recipients: A list of recipient email addresses.
"""
import requests
return requests.post(
self.endpoint,
auth=('api', self.api_key),
data={
'from': '%s <%s>' % (self.name, from_address),
'to': recipients,
'subject': subject,
'text': text
})
def process_response(self, statement, session_id=None):
"""
Send the response statement as an email.
"""
subject = 'Message from %s' % (self.name)
self.send_message(
subject,
statement.text,
self.from_address,
self.recipients
)
return statementThe Mailgun adapter allows the chat bot to send emails using the Mailgun API.
# -*- coding: utf-8 -*-
from chatterbot import ChatBot
from settings import MAILGUN
'''
To use this example, create a new file called settings.py.
In settings.py define the following:
MAILGUN = {
"CONSUMER_KEY": "my-mailgun-api-key",
"API_ENDPOINT": "https://api.mailgun.net/v3/my-domain.com/messages"
}
'''
# Change these to match your own email configuration
FROM_EMAIL = "mailgun@salvius.org"
RECIPIENTS = ["gunthercx@gmail.com"]
bot = ChatBot(
"Mailgun Example Bot",
mailgun_from_address=FROM_EMAIL,
mailgun_api_key=MAILGUN["CONSUMER_KEY"],
mailgun_api_endpoint=MAILGUN["API_ENDPOINT"],
mailgun_recipients=RECIPIENTS,
input_adapter="chatterbot.input.Mailgun",
output_adapter="chatterbot.output.Mailgun",
storage_adapter="chatterbot.storage.SQLStorageAdapter",
database="../database.db"
)
# Send an example email to the address provided
response = bot.get_response("How are you?")
print("Check your inbox at ", RECIPIENTS)
