Forward bot messages received from /tracker/events route to output channel

changelog/6273.improvement.md

@@ -0,0 +1 @@
+Add `output_channel` query param to `/conversations/<conversation_id>/tracker/events` route. Forward bot messages received by route to output channel.

docs/static/spec/rasa.yml

@@ -153,11 +153,22 @@ paths:
       summary: Append events to a tracker
       description: >-
         Appends one or multiple new events to the tracker state of the conversation.
-        Any existing events will be kept and the new events will be
-        appended, updating the existing state.
+        Any existing events will be kept and the new events will be appended,
+        updating the existing state.
       parameters:
       - $ref: '#/components/parameters/conversation_id'
       - $ref: '#/components/parameters/include_events'
+      - $ref: '#/components/parameters/output_channel'
+      - in: query
+        name: execute_side_effects
+        schema:
+          type: boolean
+          default: False
+          description: >-
+            If `true`, any ``BotUttered`` event will be forwarded to the
+            channel specified in the ``output_channel`` parameter. Any
+            ``ReminderScheduled`` or ``ReminderCancelled`` event will also be
+            processed.
       requestBody:
         required: true
         content:

rasa/core/processor.py

@@ -326,7 +326,6 @@ async def handle_reminder(
         reminder_event: ReminderScheduled,
         sender_id: Text,
         output_channel: OutputChannel,
-        nlg: NaturalLanguageGenerator,
     ) -> None:
         """Handle a reminder that is triggered asynchronously."""
 
@@ -580,6 +579,19 @@ def should_predict_another_action(action_name: Text) -> bool:
 
         return action_name not in (ACTION_LISTEN_NAME, ACTION_SESSION_START_NAME)
 
+    async def execute_side_effects(
+        self,
+        events: List[Event],
+        tracker: DialogueStateTracker,
+        output_channel: OutputChannel,
+    ) -> None:
+        """Send bot messages, schedule and cancel reminders that are logged
+        in the events array."""
+
+        await self._send_bot_messages(events, tracker, output_channel)
+        await self._schedule_reminders(events, tracker, output_channel)
+        await self._cancel_reminders(events, tracker)
+
     @staticmethod
     async def _send_bot_messages(
         events: List[Event],
@@ -599,7 +611,6 @@ async def _schedule_reminders(
         events: List[Event],
         tracker: DialogueStateTracker,
         output_channel: OutputChannel,
-        nlg: NaturalLanguageGenerator,
     ) -> None:
         """Uses the scheduler to time a job to trigger the passed reminder.
 
@@ -614,7 +625,7 @@ async def _schedule_reminders(
                 self.handle_reminder,
                 "date",
                 run_date=e.trigger_date_time,
-                args=[e, tracker.sender_id, output_channel, nlg],
+                args=[e, tracker.sender_id, output_channel],
                 id=e.name,
                 replace_existing=True,
                 name=e.scheduled_job_name(tracker.sender_id),
@@ -674,9 +685,7 @@ async def _run_action(
         ):
             self._log_slots(tracker)
 
-        await self._send_bot_messages(events, tracker, output_channel)
-        await self._schedule_reminders(events, tracker, output_channel, nlg)
-        await self._cancel_reminders(events, tracker)
+        await self.execute_side_effects(events, tracker, output_channel)
 
         return self.should_predict_another_action(action.name())
 

rasa/server.py

@@ -61,6 +61,7 @@
 
 OUTPUT_CHANNEL_QUERY_KEY = "output_channel"
 USE_LATEST_INPUT_CHANNEL_AS_OUTPUT_CHANNEL = "latest"
+EXECUTE_SIDE_EFFECTS_QUERY_KEY = "execute_side_effects"
 
 
 class ErrorResponse(Exception):
@@ -497,12 +498,19 @@ async def append_events(request: Request, conversation_id: Text):
             async with app.agent.lock_store.lock(conversation_id):
                 processor = app.agent.create_processor()
                 tracker = processor.get_tracker(conversation_id)
+                output_channel = _get_output_channel(request, tracker)
                 _validate_tracker(tracker, conversation_id)
 
                 events = _get_events_from_request_body(request)
 
                 for event in events:
                     tracker.update(event, app.agent.domain)
+                if rasa.utils.endpoints.bool_arg(
+                    request, EXECUTE_SIDE_EFFECTS_QUERY_KEY, False
+                ):
+                    await processor.execute_side_effects(
+                        events, tracker, output_channel
+                    )
                 app.agent.tracker_store.save(tracker)
 
             return response.json(tracker.current_state(verbosity))

tests/core/test_processor.py

@@ -166,7 +166,7 @@ async def test_reminder_scheduled(
     default_processor.tracker_store.save(tracker)
 
     await default_processor.handle_reminder(
-        reminder, sender_id, default_channel, default_processor.nlg
+        reminder, sender_id, default_channel,
     )
 
     # retrieve the updated tracker
@@ -197,7 +197,7 @@ async def test_reminder_aborted(
 
     default_processor.tracker_store.save(tracker)
     await default_processor.handle_reminder(
-        reminder, sender_id, default_channel, default_processor.nlg
+        reminder, sender_id, default_channel,
     )
 
     # retrieve the updated tracker
@@ -243,7 +243,7 @@ async def test_reminder_cancelled_multi_user(
     for tracker in trackers:
         default_processor.tracker_store.save(tracker)
         await default_processor._schedule_reminders(
-            tracker.events, tracker, default_channel, default_processor.nlg
+            tracker.events, tracker, default_channel,
         )
     # check that the jobs were added
     assert len((await jobs.scheduler()).get_jobs()) == 2
@@ -334,7 +334,7 @@ async def test_reminder_cancelled_by_name(
 ):
     tracker = tracker_with_six_scheduled_reminders
     await default_processor._schedule_reminders(
-        tracker.events, tracker, default_channel, default_processor.nlg
+        tracker.events, tracker, default_channel,
     )
 
     # cancel the sixth reminder
@@ -350,7 +350,7 @@ async def test_reminder_cancelled_by_entities(
 ):
     tracker = tracker_with_six_scheduled_reminders
     await default_processor._schedule_reminders(
-        tracker.events, tracker, default_channel, default_processor.nlg
+        tracker.events, tracker, default_channel,
     )
 
     # cancel the fourth reminder
@@ -370,7 +370,7 @@ async def test_reminder_cancelled_by_intent(
 ):
     tracker = tracker_with_six_scheduled_reminders
     await default_processor._schedule_reminders(
-        tracker.events, tracker, default_channel, default_processor.nlg
+        tracker.events, tracker, default_channel,
     )
 
     # cancel the third, fifth, and sixth reminder
@@ -386,7 +386,7 @@ async def test_reminder_cancelled_all(
 ):
     tracker = tracker_with_six_scheduled_reminders
     await default_processor._schedule_reminders(
-        tracker.events, tracker, default_channel, default_processor.nlg
+        tracker.events, tracker, default_channel,
     )
 
     # cancel all reminders
@@ -411,7 +411,7 @@ async def test_reminder_restart(
 
     default_processor.tracker_store.save(tracker)
     await default_processor.handle_reminder(
-        reminder, sender_id, default_channel, default_processor.nlg
+        reminder, sender_id, default_channel,
     )
 
     # retrieve the updated tracker

tests/test_server.py

@@ -25,14 +25,21 @@
 import rasa.server
 from rasa.core import events, utils
 from rasa.core.agent import Agent
-from rasa.core.channels import CollectingOutputChannel, RestInput, SlackInput
+from rasa.core.channels import (
+    channel,
+    CollectingOutputChannel,
+    RestInput,
+    SlackInput,
+    CallbackInput,
+)
 from rasa.core.channels.slack import SlackBot
 from rasa.core.events import Event, UserUttered, SlotSet, BotUttered
 from rasa.core.trackers import DialogueStateTracker
 from rasa.model import unpack_model
 from rasa.nlu.constants import INTENT_NAME_KEY
 from rasa.utils.endpoints import EndpointConfig
 from rasa import utils as rasa_utils
+from tests.utilities import json_of_latest_request, latest_request
 from sanic import Sanic
 from sanic.testing import SanicTestClient
 from tests.nlu.utilities import ResponseTest
@@ -889,6 +896,40 @@ def test_push_multiple_events(rasa_app: SanicTestClient):
     assert tracker.get("events") == events
 
 
+@pytest.mark.parametrize(
+    "params", ["?execute_side_effects=true&output_channel=callback", ""],
+)
+def test_pushing_event_while_executing_side_effects(rasa_server: Sanic, params: Text):
+    input_channel = CallbackInput(EndpointConfig("https://example.com/callback"))
+    channel.register([input_channel], rasa_server, "/webhooks/")
+    rasa_app = get_test_client(rasa_server)
+    sender_id = str(uuid.uuid1())
+    conversation = f"/conversations/{sender_id}"
+
+    serialized_event = test_events[1].as_dict()
+
+    with aioresponses() as mocked:
+        mocked.post(
+            "https://example.com/callback",
+            repeat=True,
+            headers={"Content-Type": "application/json"},
+        )
+        rasa_app.post(
+            f"{conversation}/tracker/events{params}",
+            json=serialized_event,
+            headers={"Content-Type": rasa.server.JSON_CONTENT_TYPE},
+        )
+
+        r = latest_request(mocked, "post", "https://example.com/callback")
+
+        if not params:
+            assert r is None
+        else:
+            message_received = json_of_latest_request(r)
+            assert message_received.get("recipient_id") == sender_id
+            assert message_received.get("text") == serialized_event.get("text")
+
+
 def test_post_conversation_id_with_slash(rasa_app: SanicTestClient):
     conversation_id = str(uuid.uuid1())
     id_len = len(conversation_id) // 2