# 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/existing-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.9.2/api/events/#events) from the Rasa server to other services.

- [Format](https://legacy-docs-v1.rasa.com/1.9.2/api/event-brokers/#format)
- [Pika Event Broker](https://legacy-docs-v1.rasa.com/1.9.2/api/event-brokers/#pika-event-broker)
- [Kafka Event Broker](https://legacy-docs-v1.rasa.com/1.9.2/api/event-brokers/#kafka-event-broker)
- [SQL Event Broker](https://legacy-docs-v1.rasa.com/1.9.2/api/event-brokers/#sql-event-broker)

## 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:

```json
{
    "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`.

## 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.9.2/api/event-brokers/#adding-a-pika-event-broker-using-the-endpoint-configuration)
- [Adding a Pika Event Broker in Python](https://legacy-docs-v1.rasa.com/1.9.2/api/event-brokers/#adding-a-pika-event-broker-in-python)
- [Implementing a Pika Event Consumer](https://legacy-docs-v1.rasa.com/1.9.2/api/event-brokers/#implementing-a-pika-event-consumer)

### 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`:

```yaml
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

Here is how you add it using Python code:

```python
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

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:

```python
import json
import pika

def _callback(self, ch, method, properties, body):
        # Do something useful with your incoming message body here, e.g.
        # saving it to a database
        print('Received event {}'.format(json.loads(body)))

if __name__ == '__main__':
    # RabbitMQ credentials with username and password
    credentials = pika.PlainCredentials('username', 'password')

# Pika connection to the RabbitMQ host - typically 'rabbit' in a
    # docker environment, or 'localhost' in a local environment
    connection = pika.BlockingConnection(
        pika.ConnectionParameters('rabbit', credentials=credentials))

# start consumption of channel
    channel = connection.channel()
    channel.basic_consume(_callback,
                          queue='rasa_events',
                          no_ack=True)
    channel.start_consuming()
```

## Kafka Event Broker

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

- [Adding a Kafka Event Broker Using the Endpoint Configuration](https://legacy-docs-v1.rasa.com/1.9.2/api/event-brokers/#adding-a-kafka-event-broker-using-the-endpoint-configuration)
- [Adding a Kafka Broker in Python](https://legacy-docs-v1.rasa.com/1.9.2/api/event-brokers/#adding-a-kafka-broker-in-python)
- [Authentication and Authorization](https://legacy-docs-v1.rasa.com/1.9.2/api/event-brokers/#authentication-and-authorization)
- [Implementing a Kafka Event Consumer](https://legacy-docs-v1.rasa.com/1.9.2/api/event-brokers/#implementing-a-kafka-event-consumer)

### 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`.

Using `SASL_PLAINTEXT` protocol the endpoints file must have the following entries:

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

If using SSL protocol, the endpoints file should look like:

```yaml
event_broker:
  url: localhost
  topic: topic
  security_protocol: SSL
  ssl_cafile: CARoot.pem
  ssl_certfile: certificate.pem
  ssl_keyfile: key.pem
  ssl_check_hostname: True
  type: kafka
```

### Adding a Kafka Broker in Python

The code below shows an example on how to instantiate a Kafka producer in your script.

```python
from rasa.core.brokers.kafka import KafkaEventBroker
from rasa.core.tracker_store import InMemoryTrackerStore

kafka_broker = KafkaEventBroker(host='localhost:9092',
                                topic='rasa_events')

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

### Authentication and Authorization

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

For a development environment, or if the brokers servers and clients are located on the same machine, you can use simple authentication with `SASL_PLAINTEXT`. By using this protocol, the credentials and messages exchanged between the clients and servers will be sent in plaintext. Thus, this is not the most secure approach, but since it’s simple to configure, it is useful for simple cluster configurations.

```python
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 or the brokers in the kafka cluster are located in different machines, it’s important to use SSL protocol to ensure encryption of data and client authentication. After generating valid certificates for the brokers and the clients, the path to the certificate and key generated for the producer must be provided as arguments, as well as the CA’s root certificate.

```python
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

The parameters to create a Kafka consumer are the same as those used in the producer creation, according to the security protocol being used. The following implementation shows an example:

```python
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

It is possible to use an SQL database as an event broker. Connections to databases are established using [SQLAlchemy](https://www.sqlalchemy.org/), a Python library that can interact with many different types of SQL databases, such as [SQLite](https://sqlite.org/) and [PostgreSQL](https://www.postgresql.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`. For example, a valid SQLite configuration could look like the following:

```yaml
event_broker:
  type: SQL
  dialect: sqlite
  db: events.db
```
  
PostgreSQL databases can be used as well:

```yaml
event_broker:
  type: SQL
  host: 127.0.0.1
  port: 5432
  dialect: postgresql
  username: myuser
  password: mypassword
  db: mydatabase
```

With this configuration applied, Rasa will create a table called `events` in the database, where all events will be added.