# Event Brokers

An event broker allows you to connect your running assistant to other services that process the data coming in from conversations. For example, you could [connect your live assistant to Rasa X](/content/docs/rasa-x/installation-and-setup/deploy#connect-rasa-deployment/index.html) to review and annotate conversations or forward messages to an external analytics service. The event broker publishes messages to a message streaming service, also known as a message broker, to forward Rasa [Events](https://legacy-docs-v1.rasa.com/1.10.9/api/events/#events) from the Rasa server to other services.

## [Format](https://legacy-docs-v1.rasa.com/1.10.9/api/event-brokers/#format)

All events are streamed to the broker as serialized dictionaries every time the tracker updates its state. An example event emitted from the `default` tracker looks like this:

```
{
    "sender_id": "default",
    "timestamp": 1528402837.617099,
    "event": "bot",
    "text": "what your bot said",
    "data": "some data about e.g. attachments"
    "metadata": {
          "a key": "a value"
     }
}
```

The `event` field takes the event’s `type_name` (for more on event types, check out the [Events](https://legacy-docs-v1.rasa.com/1.10.9/api/events/#events) docs).

## [Pika Event Broker](https://legacy-docs-v1.rasa.com/1.10.9/api/event-brokers/#pika-event-broker)

The example implementation we’re going to show you here uses [Pika](https://pika.readthedocs.io/), the Python client library for [RabbitMQ](https://www.rabbitmq.com/).

### [Adding a Pika Event Broker Using the Endpoint Configuration](https://legacy-docs-v1.rasa.com/1.10.9/api/event-brokers/#adding-a-pika-event-broker-using-the-endpoint-configuration)

You can instruct Rasa to stream all events to your Pika event broker by adding an `event_broker` section to your `endpoints.yml`:

```
event_broker:
  type: pika
  url: localhost
  username: username
  password: password
  queues:
    - queue-1
#   you may supply more than one queue to publish to
#   - queue-2
#   - queue-3
```

Rasa will automatically start streaming events when you restart the Rasa server.

### [Adding a Pika Event Broker in Python](https://legacy-docs-v1.rasa.com/1.10.9/api/event-brokers/#adding-a-pika-event-broker-in-python)

Here is how you add it using Python code:

```
from rasa.core.brokers.pika import PikaEventBroker
from rasa.core.tracker_store import InMemoryTrackerStore

pika_broker = PikaEventBroker('localhost',
                              'username',
                              'password',
                              queues=['rasa_events'])

tracker_store = InMemoryTrackerStore(domain=domain, event_broker=pika_broker)
```

### [Implementing a Pika Event Consumer](https://legacy-docs-v1.rasa.com/1.10.9/api/event-brokers/#implementing-a-pika-event-consumer)

You need to have a RabbitMQ server running, as well as another application that consumes the events. This consumer needs to implement Pika’s `start_consuming()` method with a `callback` action. Here’s a simple example:

```
import json
import pika

def _callback(self, ch, method, properties, body):
        print('Received event {}'.format(json.loads(body)))

if __name__ == '__main__':
    credentials = pika.PlainCredentials('username', 'password')
    connection = pika.BlockingConnection(
        pika.ConnectionParameters('rabbit', credentials=credentials))
    channel = connection.channel()
    channel.basic_consume(_callback,
                          queue='rasa_events',
                          no_ack=True)
    channel.start_consuming()
```

## [Kafka Event Broker](https://legacy-docs-v1.rasa.com/1.10.9/api/event-brokers/#kafka-event-broker)

It is possible to use [Kafka](https://kafka.apache.org/) as the main broker for your events. In this example, we will use the [python-kafka](https://kafka-python.readthedocs.io/en/master/usage.html) library.

### [Adding a Kafka Event Broker Using the Endpoint Configuration](https://legacy-docs-v1.rasa.com/1.10.9/api/event-brokers/#adding-a-kafka-event-broker-using-the-endpoint-configuration)

You can instruct Rasa to stream all events to your Kafka event broker by adding an `event_broker` section to your `endpoints.yml`:

```
event_broker:
  url: localhost
  sasl_username: username
  sasl_password: password
  topic: topic
  security_protocol: SASL_PLAINTEXT
  type: kafka
```

### [Authentication and Authorization](https://legacy-docs-v1.rasa.com/1.10.9/api/event-brokers/#authentication-and-authorization)

Rasa’s Kafka producer accepts two types of security protocols - `SASL_PLAINTEXT` and `SSL`.

For development environment, you can use simple authentication with `SASL_PLAINTEXT`. By using this protocol, the credentials exchanged will be sent in plaintext.

```
kafka_broker = KafkaEventBroker(host='kafka_broker:9092',
                                sasl_plain_username='kafka_username',
                                sasl_plain_password='kafka_password',
                                security_protocol='SASL_PLAINTEXT',
                                topic='rasa_events')
```

If the clients in the Kafka cluster are located in different machines, it’s recommended to use SSL protocol to ensure the encryption of data and client authentication. After generating valid certificates, the path to the certificate and key must be provided:

```
kafka_broker = KafkaEventBroker(host='kafka_broker:9092',
                                ssl_cafile='CARoot.pem',
                                ssl_certfile='certificate.pem',
                                ssl_keyfile='key.pem',
                                ssl_check_hostname=True,
                                security_protocol='SSL',
                                topic='rasa_events')
```

### [Implementing a Kafka Event Consumer](https://legacy-docs-v1.rasa.com/1.10.9/api/event-brokers/#implementing-a-kafka-event-consumer)

The parameters used to create a Kafka consumer are the same used on the producer based on the security protocol:

```
from kafka import KafkaConsumer
from json import loads

consumer = KafkaConsumer('rasa_events',
                          bootstrap_servers=['localhost:29093'],
                          value_deserializer=lambda m: json.loads(m.decode('utf-8')), 
                          security_protocol='SSL',
                          ssl_check_hostname=False,
                          ssl_cafile='CARoot.pem',
                          ssl_certfile='certificate.pem',
                          ssl_keyfile='key.pem')

for message in consumer:
    print(message.value)
```

## [SQL Event Broker](https://legacy-docs-v1.rasa.com/1.10.9/api/event-brokers/#sql-event-broker)

It is possible to use an SQL database as an event broker. Connections to databases are established using [SQLAlchemy](https://www.sqlalchemy.org/).

### Adding a SQL Event Broker Using the Endpoint Configuration

To instruct Rasa to save all events to your SQL event broker, add an `event_broker` section to your `endpoints.yml`. An example SQLite configuration:

```
event_broker:
  type: SQL
  dialect: sqlite
  db: events.db
```