Return 449 status code if the Domain is not stored or missing in the …

rasa_sdk/endpoint.py

@@ -31,7 +31,11 @@
  from rasa_sdk.cli.arguments import add_endpoint_arguments
  from rasa_sdk.constants import DEFAULT_KEEP_ALIVE_TIMEOUT, DEFAULT_SERVER_PORT
  from rasa_sdk.executor import ActionExecutor
- from rasa_sdk.interfaces import ActionExecutionRejection, ActionNotFoundException
+ from rasa_sdk.interfaces import (
+ ActionExecutionRejection,
+ ActionNotFoundException,
+ ActionMissingDomainException,
+ )
  from rasa_sdk.plugin import plugin_manager
  from rasa_sdk.tracing.utils import get_tracer_and_context, set_span_attributes
 
@@ -139,6 +143,10 @@ async def webhook(request: Request) -> HTTPResponse:
  logger.error(e)
  body = {"error": e.message, "action_name": e.action_name}
  return response.json(body, status=404)
+ except ActionMissingDomainException as e:
+ logger.error(e)
+ body = {"error": e.message, "action_name": e.action_name}
+ return response.json(body, status=449)
 
  set_span_attributes(span, action_call)
 

rasa_sdk/executor.py

@@ -2,29 +2,29 @@
 import inspect
 import logging
 import pkgutil
-import typing
 import warnings
 from typing import Text, List, Dict, Any, Type, Union, Callable, Optional, Set, cast
 from collections import namedtuple
 import types
 import sys
 import os
 
-from rasa_sdk.interfaces import Tracker, ActionNotFoundException, Action
+from rasa_sdk.interfaces import (
+ Tracker,
+ ActionNotFoundException,
+ Action,
+ ActionMissingDomainException,
+)
 
 from rasa_sdk import utils
 
-if typing.TYPE_CHECKING: # pragma: no cover
- from rasa_sdk.types import ActionCall
-
 logger = logging.getLogger(__name__)
 
 
 class CollectingDispatcher:
  """Send messages back to user"""
 
  def __init__(self) -> None:
-
  self.messages: List[Dict[Text, Any]] = []
 
  def utter_message(
@@ -162,6 +162,8 @@ def __init__(self) -> None:
  self.actions: Dict[Text, Callable] = {}
  self._modules: Dict[Text, TimestampModule] = {}
  self._loaded: Set[Type[Action]] = set()
+ self.domain: Optional[Dict[Text, Any]] = None
+ self.domain_digest: Optional[Text] = None
 
  def register_action(self, action: Union[Type[Action], Action]) -> None:
  if inspect.isclass(action):
@@ -380,7 +382,57 @@ def validate_events(events: List[Dict[Text, Any]], action_name: Text):
  # we won't append this to validated events -> will be ignored
  return validated
 
- async def run(self, action_call: "ActionCall") -> Optional[Dict[Text, Any]]:
+ def is_domain_digest_valid(self, domain_digest: Optional[Text]) -> bool:
+ """Check if the domain_digest is valid
+
+ If the domain_digest is empty or different from the one provided, it is invalid.
+
+ Args:
+ domain_digest: latest value provided to compare the current value with.
+
+ Returns:
+ True if the domain_digest is valid, False otherwise.
+ """
+ return bool(self.domain_digest) and self.domain_digest == domain_digest
+
+ def update_and_return_domain(
+ self, payload: Dict[Text, Any], action_name: Text
+ ) -> Optional[Dict[Text, Any]]:
+ """Validate the digest, store the domain if available, and return the domain.
+
+ This method validates the domain digest from the payload.
+ If the digest is invalid and no domain is provided, an exception is raised.
+ If domain data is available, it stores the domain and digest.
+ Finally, it returns the domain.
+
+ Args:
+ payload: Request payload containing the domain data.
+ action_name: Name of the action that should be executed.
+
+ Returns:
+ The domain dictionary.
+
+ Raises:
+ ActionMissingDomainException: Invalid digest and no domain data available.
+
+ """
+ payload_domain = payload.get("domain")
+ payload_domain_digest = payload.get("domain_digest")
+
+ # If digest is invalid and no domain is available - raise the error
+ if (
+ not self.is_domain_digest_valid(payload_domain_digest)
+ and payload_domain is None
+ ):
+ raise ActionMissingDomainException(action_name)
+
+ if payload_domain:
+ self.domain = payload_domain
+ self.domain_digest = payload_domain_digest
+
+ return self.domain
+
+ async def run(self, action_call: Dict[Text, Any]) -> Optional[Dict[Text, Any]]:
  from rasa_sdk.interfaces import Tracker
 
  action_name = action_call.get("next_action")
@@ -391,7 +443,7 @@ async def run(self, action_call: "ActionCall") -> Optional[Dict[Text, Any]]:
  raise ActionNotFoundException(action_name)
 
  tracker_json = action_call["tracker"]
- domain = action_call.get("domain", {})
+ domain = self.update_and_return_domain(action_call, action_name)
  tracker = Tracker.from_dict(tracker_json)
  dispatcher = CollectingDispatcher()
 

rasa_sdk/interfaces.py

@@ -384,3 +384,14 @@ def __init__(self, action_name: Text, message: Optional[Text] = None) -> None:
 
  def __str__(self) -> Text:
  return self.message
+
+
+class ActionMissingDomainException(Exception):
+ """Raising this exception when the domain is missing."""
+
+ def __init__(self, action_name: Text, message: Optional[Text] = None) -> None:
+ self.action_name = action_name
+ self.message = message or "Domain context is missing."
+
+ def __str__(self) -> Text:
+ return self.message

tests/test_endpoint.py

@@ -59,6 +59,7 @@ def test_server_webhook_handles_action_exception():
  data = {
  "next_action": "custom_action_exception",
  "tracker": {"sender_id": "1", "conversation_id": "default"},
+ "domain": {},
  }
  request, response = app.test_client.post("/webhook", data=json.dumps(data))
  assert response.status == 500
@@ -70,6 +71,7 @@ def test_server_webhook_custom_action_returns_200():
  data = {
  "next_action": "custom_action",
  "tracker": {"sender_id": "1", "conversation_id": "default"},
+ "domain": {},
  }
  request, response = app.test_client.post("/webhook", data=json.dumps(data))
  events = response.json.get("events")
@@ -82,6 +84,7 @@ def test_server_webhook_custom_async_action_returns_200():
  data = {
  "next_action": "custom_async_action",
  "tracker": {"sender_id": "1", "conversation_id": "default"},
+ "domain": {},
  }
  request, response = app.test_client.post("/webhook", data=json.dumps(data))
  events = response.json.get("events")
@@ -140,6 +143,7 @@ def test_server_webhook_custom_action_with_dialogue_stack_returns_200(
  data = {
  "next_action": "custom_action_with_dialogue_stack",
  "tracker": {"sender_id": "1", "conversation_id": "default", **stack_state},
+ "domain": {},
  }
  _, response = app.test_client.post("/webhook", data=json.dumps(data))
  events = response.json.get("events")

tests/tracing/instrumentation/test_tracing.py

@@ -39,6 +39,7 @@ def test_server_webhook_custom_action_is_instrumented(
  """Tests that the custom action is instrumented."""
 
  data["next_action"] = action_name
+ data["domain"] = {}
  app = ep.create_app(action_package, tracer_provider=tracer_provider)
  _, response = app.test_client.post("/webhook", data=json.dumps(data))