Generated with pydoc-markdown

Shared

shared.tracker

shared.tracker.apis.feedback_processing_api

shared.tracker.apis.authorized_api

shared.tracker.apis._statistics_api

StatisticsApi Objects​

class StatisticsApi(StateManipulationApi)

API for emitting custom statistics events.

Custom statistics are stored as tracker events and aggregated in the analytics dashboard.

add_custom_statistic​

def add_custom_statistic(name: str, value: int = 1)

Increment a custom statistic by a given value.

Arguments:

• name str - Name of the statistic.
• value int, optional - Value to increment by. Defaults to 1.

Raises:

• ValueError - If value is less than 1.

shared.tracker.apis._utter_api

ActionServiceManagerConfig Objects​

class ActionServiceManagerConfig(BaseModel)

Configuration for ActionsManager service endpoints.

UtterApiConfig Objects​

class UtterApiConfig(StateManipulationApiConfig, FunctionExecutionApiConfig,
                     TextNormalizeApiConfig)

Configuration for utterance APIs with optional postprocessing.

UtterApi Objects​

class UtterApi(StateManipulationApi, FunctionExecutionApi, TextNormalizeApi)

API for generating bot responses and utterances.

Provides methods for LLM-based responses, fixed utterances, streaming responses, and postprocessing of generated content.

llm_utter​

async def llm_utter(chat_api_response: ChatApiResponse,
                    metadata: dict[str, Any] | None = None,
                    buttons: list[Any] | None = None,
                    image: str | None = None,
                    video: str | None = None,
                    iFrame: IFrameDefinition | None = None,
                    execute_post_processing: bool = True,
                    sources: list[str] | None = None,
                    feedback_enabled: bool | None = None,
                    feedback_metadata: dict[str, Any] | None = None)

Send an utterance, generated by the LLM to the user. Use this for LLM Utterances to persist the inference parameters used to generate this utterance in the history.

Arguments:

• chat_api_response ChatApiResponse - The response from the LLM API Endpoint.
• metadata dict[str, Any], optional - Optional Metadata that should be persisted in the history. Defaults to {}.
• buttons Optional[list[Any]], optional - list of buttons to be sent to the user. Defaults to [].
• image Optional[str], optional - Image URL to be sent to the user. Defaults to None.
• video Optional[str], optional - Video URL to be sent to the user. Defaults to None.
• iFrame Optional[IFrameDefinition], optional - iFrame to be sent to the user. Defaults to None.
• execute_post_processing bool, optional - Whether to execute the post processing script. Defaults to True.
• sources list[str], optional - Optional list of source citations as Markdown strings, shown in a collapsible section under the bot response in the frontend. Defaults to None.
• feedback_enabled bool | None, optional - If set, overrides node feedback for this utterance. If None, uses the tracker default from node/bot config. Defaults to None.
• feedback_metadata dict[str, Any] | None, optional - Arbitrary metadata to carry along with this utterance for later feedback handling. If None, defaults to {}.

fixed_utter​

def fixed_utter(text: str,
                image: str | None = None,
                video: str | None = None,
                buttons: list[Any] | None = None,
                iFrame: IFrameDefinition | None = None,
                metadata: dict[str, Any] | None = None,
                use_normalize: bool | None = None,
                sources: list[str] | None = None,
                feedback_enabled: bool | None = None,
                feedback_metadata: dict[str, Any] | None = None)

Fixed Utterance, this is used to send a fixed utterance to the user. Text will be uttered as soon as this method is called.

Arguments:

• text str - Text to be sent to the user
• image Optional[str], optional - Image URL to be sent to the user. Defaults to None.
• video Optional[str], optional - Video URL to be sent to the user. Defaults to None.
• buttons Optional[list[Any]], optional - list of buttons to be sent to the user. Defaults to [].
• iFrame Optional[IFrameDefinition], optional - iFrame to be sent to the user. Defaults to None.
• metadata Optional[dict[str, Any]], optional - Additional metadata to be sent to the user. Defaults to None.
• use_normalize Optional[bool], optional - Override text normalization for this call. If True, forces normalization. If False, disables it. If None, uses system and node settings. Defaults to None.
• sources list[str], optional - Optional list of source citations as Markdown strings, shown in a collapsible section under the bot response in the frontend. Defaults to None.
• feedback_enabled bool | None, optional - If set, overrides node feedback for this utterance. If None, uses the tracker default from node/bot config. Defaults to None.
• feedback_metadata dict[str, Any] | None, optional - Arbitrary metadata to carry along with this utterance for later feedback handling. If None, defaults to {}.

stream_response_model​

async def stream_response_model(
        streaming_api_response: StreamingChatApiRequest,
        stream_key: str,
        accept_flag: str | None = None,
        metadata: dict[str, Any] | None = None,
        buttons: list[Any] | None = None,
        image: str | None = None,
        video: str | None = None,
        iFrame: IFrameDefinition | None = None,
        sources: list[str] | None = None,
        feedback_enabled: bool | None = None,
        feedback_metadata: dict[str, Any] | None = None) -> bool

Stream a structured output (response model) call to the user, uttering only the value of a specific key from the JSON response.

The LLM is expected to return valid JSON matching the provided response_model. As the JSON streams in, only the value of stream_key is extracted and streamed to the user as a chat utterance.

If accept_flag is provided, it names a boolean field in the response model that must be true for streaming to occur. If the flag resolves to false, no events are emitted and the method returns False after the background LLM request completes (so the LLMInferenceEvent is still fully populated). The accept_flag field must appear before stream_key in the response model's field order.

Note: This method does not validate the full JSON against the response model. It is a streaming function only. If you need the parsed model, use a_call_llm_with_response_model instead.

Important: Node post-processing scripts are currently not applied in this method. Unlike stream_utter, this method does not execute postProcessScript handlers (default, stream, or advanced_stream) and emits the extracted stream_key value directly.

Code Example:

class AnswerTemplate(BaseModel):
    reasoning: str
    answerable_from_available_information: bool
    concise_answer: str


stream = await tracker_api.a_stream_call_response_model(
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": tracker_api.get_latest_user_message().content},
    ],
    response_model=AnswerTemplate,
)
streamed = await tracker_api.stream_response_model(
    stream,
    stream_key="concise_answer",
    accept_flag="answerable_from_available_information",
)
if not streamed:
    await tracker_api.fixed_utter("I can't answer that from the available information.")

Arguments:

• streaming_api_response StreamingChatApiRequest - The streaming response from a_stream_call_response_model.
• stream_key str - The JSON key whose value should be streamed as an utterance to the user.
• accept_flag str, optional - A boolean key in the response model. If provided and the value resolves to false, streaming is skipped and False is returned. Defaults to None.
• metadata Dict[str, Any], optional - Optional metadata. Defaults to None.
• buttons list[Any], optional - Optional buttons. Defaults to None.
• image str, optional - Optional image. Defaults to None.
• video str, optional - Optional video. Defaults to None.
• iFrame IFrameDefinition, optional - Optional iFrame. Defaults to None.
• sources list[str] | None, optional - Optional list of source citations as Markdown strings, shown in a collapsible section under the bot response in the frontend. Defaults to None.
• feedback_enabled bool | None, optional - If set, overrides node feedback for this utterance. If None, uses the tracker default from node/bot config. Defaults to None.
• feedback_metadata dict[str, Any] | None, optional - Arbitrary metadata to carry along with this utterance for later feedback handling. If None, defaults to {}.

Returns:

• bool - True if the stream_key value was streamed to the user, False if accept_flag was false and streaming was skipped.

a_stream_utter_and_extract_tool_calls​

async def a_stream_utter_and_extract_tool_calls(
        streaming_api_response: StreamingChatApiRequest,
        metadata: dict[str, Any] | None = None,
        buttons: list[Any] | None = None,
        image: str | None = None,
        video: str | None = None,
        iFrame: IFrameDefinition | None = None,
        sources: list[str] | None = None,
        feedback_enabled: bool | None = None,
        feedback_metadata: dict[str, Any] | None = None
) -> list[dict[str, Any]]

Stream the LLM response to the user and extract all tool calls from the same stream. Text is streamed to the user while tool call chunks are collected in the background; the full tool calls (including parallel tool calls) are returned after the stream ends, sorted by their streamed index. Returns an empty list if no tool call was present.

Arguments:

• streaming_api_response - The streaming request.
• metadata - Optional metadata for the uttered message.
• buttons - Optional buttons.
• image - Optional image.
• video - Optional video.
• iFrame - Optional iFrame definition.
• sources - Optional list of source citations as Markdown strings, shown in a collapsible section under the bot response in the frontend. Defaults to None.
• feedback_enabled - If set, overrides node feedback for this utterance. If None, uses the tracker default from node/bot config. Defaults to None.
• feedback_metadata - Arbitrary metadata to carry along with this utterance for later feedback handling. If None, defaults to {}.

Returns:

List of tool call dicts sorted by index, empty if no tool call was present.

a_stream_utter_and_extract_tool_call​

async def a_stream_utter_and_extract_tool_call(
        streaming_api_response: StreamingChatApiRequest,
        metadata: dict[str, Any] | None = None,
        buttons: list[Any] | None = None,
        image: str | None = None,
        video: str | None = None,
        iFrame: IFrameDefinition | None = None,
        sources: list[str] | None = None,
        feedback_enabled: bool | None = None,
        feedback_metadata: dict[str, Any] | None = None
) -> dict[str, Any] | None

Stream the LLM response to the user and extract a single tool call from the same stream. Text is streamed to the user while tool call chunks are collected in the background; the full tool call is returned after the stream ends, or None if no tool call was present.

If the LLM makes multiple parallel tool calls, only the first one (by index) is returned and a warning is logged. Use a_stream_utter_and_extract_tool_calls to receive all of them.

Arguments:

• streaming_api_response - The streaming request.
• metadata - Optional metadata for the uttered message.
• buttons - Optional buttons.
• image - Optional image.
• video - Optional video.
• iFrame - Optional iFrame definition.
• sources - Optional list of source citations as Markdown strings, shown in a collapsible section under the bot response in the frontend. Defaults to None.
• feedback_enabled - If set, overrides node feedback for this utterance. If None, uses the tracker default from node/bot config. Defaults to None.
• feedback_metadata - Arbitrary metadata to carry along with this utterance for later feedback handling. If None, defaults to {}.

Returns:

Tool call dict or None.

shared.tracker.apis.flow_api

shared.tracker.apis.config_map_api

shared.tracker.apis.legacy_apis

shared.tracker.apis._base_assistant_api

BaseAssistantApiConfig Objects​

class BaseAssistantApiConfig(ConfigMapApiConfig)

Base configuration for assistant APIs.

BaseAssistantApi Objects​

class BaseAssistantApi(ConfigMapApi)

Base API providing core assistant capabilities.

Provides methods for configuration access, validation execution, and JSON event emission.

get_visual_tag​

def get_visual_tag() -> str | None

Get the visual tag of the bot. The visual tag indicates the bot's deployment stage.

Returns:

str | None: The visual tag (e.g., "dev", "staging", "prod") or None if not set.

get_config_value​

def get_config_value(key: str) -> str | None

Get a configuration value from the bot's configmap.

Arguments:

• key - The configuration key to retrieve

Returns:

The configuration value as a string, or None if the key doesn't exist

get_bot_persona​

def get_bot_persona() -> str | None

Get the bot persona description from configuration.

Returns:

str | None: The bot persona text, or None if not configured.

yield_json_event​

async def yield_json_event(metadata: dict[str, Any], test_only: bool = False)

Yield a JSON Event. The JSON Event does not influence the Engines behaviour in any way. This methods purpose is solely to use the "metadata" attribute of a base EventPayload to yield custom data from the Chat endpoint. Yields the given data in event.payload.metadata to the client.

Arguments:

• metadata Dict[str, Any] - The metadata that should be included in the JSON Event
• test_only bool - If True, the event will only be yielded to the test client. Defaults to False.

shared.tracker.apis.periodic_script_api

shared.tracker.apis._tag_api

TagApiConfig Objects​

class TagApiConfig(StateManipulationApiConfig)

Configuration for tag API containing current tag values.

tags​

List of tag names currently assigned to this tracker

TagManipulationApi Objects​

class TagManipulationApi(StateManipulationApi)

API for manipulating conversation tags.

Only available in RemoteLLMTrackerApi (in-story execution).

Tags are used for categorizing chat history for filtering and analysis. Tags must be pre-created in the system before they can be assigned.

add_tag​

def add_tag(tag_name: str)

Add a tag to the current conversation by name.

If the tag doesn't exist, it will be created automatically with a default color.

Arguments:

• tag_name str - Name of the tag to add to this conversation.

Raises:

• ValueError - If tag_name is empty.

Example:

tracker_api.add_tag("complaint") tracker_api.add_tag("escalation")

get_tags​

@StateManipulationApi.guarded_state
def get_tags() -> list[str]

Get all tags assigned to the conversation.

Returns:

• list[str] - List of tag names.

shared.tracker.apis.text_normalize_api

shared.tracker.apis.base_assistant_api

shared.tracker.apis.phone_api

shared.tracker.apis.llm_base_api

shared.tracker.apis.conversation_snapshot_api

shared.tracker.apis._feedback_processing_api

FeedbackProcessingTrackerApiConfig Objects​

class FeedbackProcessingTrackerApiConfig(LLMCallApiConfig, TagApiConfig,
                                         ConfigMapApiConfig)

Configuration for feedback processing tracker API.

Combines the standard LLM call configuration (bot/session context, providers, configmap) with the tag API state plus the public conversation state snapshot used in feedback scripts.

FeedbackProcessingTrackerApi Objects​

class FeedbackProcessingTrackerApi(LLMCallApi, TagManipulationApi,
                                   StatisticsApi, ConfigMapApi,
                                   ConversationSnapshotApi)

Tracker API for feedback processing scripts.

Provides:

• LLM calling helpers from LLMCallApi (a_call_llm, a_call_llm_with_response_model, etc.)
• Tag manipulation helpers from TagManipulationApi (add_tag, get_tags)
• Custom statistics via StatisticsApi (add_custom_statistic)
• latest_user_msg, history, slots as direct attributes from the conversation state snapshot
• Convenience accessors consistent with the tracker API surface

bot_id​

@property
def bot_id() -> str

Bot ID of the current session.

tenant_id​

@property
def tenant_id() -> str

Tenant ID of the current session.

session_id​

@property
def session_id() -> str

Session ID of the current conversation.

shared.tracker.apis.slot_api

shared.tracker.apis.base

shared.tracker.apis.llm_call_api

shared.tracker.apis.utter_api

shared.tracker.apis._llm_base_api

LLMBaseApiConfig Objects​

class LLMBaseApiConfig(StateManipulationApiConfig)

Configuration for LLM base APIs.

LLMBaseApi Objects​

class LLMBaseApi(StateManipulationApi)

API for accessing conversation history and LLM parameters.

Provides methods to get/set LLM parameters, access conversation history, and retrieve user messages.

get_llm_parameters​

async def get_llm_parameters() -> LLMBaseParameters

Get the LLM Parameters that are currently set for the llm inference in the frontend.

Returns:

• LLMBaseParameters - The LLM Parameters that are currently set in the LLM API Config via the frontend.

set_llm_parameters​

async def set_llm_parameters(llm_parameters: LLMBaseParameters)

Set the LLM Parameters for the llm inference in the frontend.

Arguments:

• llm_parameters LLMBaseParameters - The LLM Parameters that should be set in the LLM API Config via the frontend.

a_get_history​

async def a_get_history(
        n_turns: int | None = None,
        unsafe: bool = False,
        include_livechat_history: bool = False) -> list[ConversationTurnType]

Get the history of the conversation, optionally including livechat sessions. Synchronizes the state from the backend before returning the history.

Arguments:

• n_turns Optional[int], optional - Number of UserTurns to return (excluding the current User Input). If 0 returns an empty list, if > 0 returns up to n UserTurns starting from the back, if smaller 0 returns the complete history. If None, the history attribute of llm_api_config is used instead. When include_livechat_history=True, counts both UserTurn and LivechatUserTurn.
• unsafe bool - If True, the history is returned unsanitized (e.g. with the raw text of the user turn not sanitized by the guard agent).
• include_livechat_history bool - If True, includes parsed livechat sessions merged with conversation history, sorted by timestamp. Only available when the API has access to _get_livechat_sessions method.

Returns:

list[ConversationTurnType | LivechatTurnBase]: The conversation history, optionally merged with livechat history.

get_history​

def get_history(
        n_turns: int | None = None,
        unsafe: bool = False,
        include_livechat_history: bool = False) -> list[ConversationTurnType]

Get the conversation history. This method returns the conversation history as a list of UserTurn and AssistantTurn objects.

Arguments:

• n_turns Optional[int], optional - Number of UserTurns to return (excluding the current User Input). If 0 returns an empty list, if > 0 returns up to n UserTurns starting from the back, if smaller 0 returns the complete history.
• unsafe bool - If True, the history is returned unsanitized (e.g. with the raw text of the user turn not sanitized by the guard agent).
• include_livechat_history bool - If True, raises an error. Use a_get_history() instead for livechat support.

Returns:

list[UserTurn | AssistantTurn | ToolTurn | SystemTurn | LivechatOperatorTurn | LivechatUserTurn]: The conversation history

Raises:

• ValueError - If include_livechat_history=True (not supported in sync method)

reset_chat_history​

def reset_chat_history()

Reset the chat history.

get_latest_user_message​

def get_latest_user_message() -> UserTurn

Get the latest user message as a UserTurn object

Returns:

• UserTurn - The latest user message

history​

@property
@StateManipulationApi.guarded_state
def history() -> list[ConversationTurnType]

Get the history of the conversation. This method returns the history of the conversation. The history is returned as a list of ConversationTurnType objects. .. deprecated:: This method is deprecated and will return unsynchronized state values. To make sure you get the latest state values, please use the async method a_history.

Returns:

• list[ConversationTurnType] - The history of the conversation.

a_history​

@property
async def a_history() -> list[ConversationTurnType]

Get the history of the conversation. This method returns the history of the conversation. The history is returned as a list of ConversationTurnType objects.

Returns:

• list[ConversationTurnType] - The history of the conversation.

shared.tracker.apis._flow_api

PreStoryFlowApiConfig Objects​

class PreStoryFlowApiConfig(StateManipulationApiConfig)

Configuration for pre-story flow API with read-only flow state information.

FlowApiConfig Objects​

class FlowApiConfig(PreStoryFlowApiConfig)

Configuration for in-story flow API with guaranteed non-null flow state.

FlowApi Objects​

class FlowApi(StatisticsApi)

API for reading flow state and performing basic flow operations.

Provides methods to access story state, reset stories, and trigger livechat handovers. Custom statistics are available via StatisticsApi.

reset_story_state​

def reset_story_state(reset_slots: list[str] | None = None,
                      keep_slots: list[str] | None = None,
                      reset_all_slots: bool = False,
                      **kwargs: dict[str, Any])

Reset the story state. This method is used to reset the story state. This will reset the currently active stories state to the initial state. The next user utterance will therefore be send to the fist step of the story. By default this will not reset all Slots of the Story. You can specify which slots should be reset.

The slots are reset based on the following priority:

1. If reset_all_slots is True or keep_slots is not empty, all slots will be reset except those in keep_slots
2. Otherwise, only slots listed in reset_slots will be reset while keeping all others

Arguments:

• reset_slots Set[str] - Set of slot names that should be reset if no other reset method is active, formerly named slots
• keep_slots Set[str] - Set of slot names that should be kept
• reset_all_slots bool - Flag to reset all slots except those in keep_slots

Notes:

The keep_slots parameter only has an effect when reset_all_slots is True or keep_slots is not empty. In these cases, all slots NOT in keep_slots will be reset. The reset_slots parameter is only used when reset_all_slots is False and keep_slots is empty.

reset_bot​

def reset_bot(next_story_name: str | None = None) -> None

Reset the complete bot tracker state.

Reset scope:

• orchestrator_history
• slots
• history
• latest_user_msg
• num_nodes_since_last_user_input
• story_states (reset to start nodes)
• error_flow_story_id

By default, the bot requests fresh user input. Optionally, a target flow can be provided to continue execution immediately.

Arguments:

• next_story_name - Optional flow name to start after reset. If None, the bot waits for user input.

reset_story_state_by_name​

def reset_story_state_by_name(story_name: str)

Reset the story state by name.

get_active_node_story​

def get_active_node_story() -> str | None

Get the story name for the active node. (Doesn't matter if it's a substory or not)

Returns:

• Optional[str] - The story name for the active node.

get_orchestrator_history​

def get_orchestrator_history() -> list[str]

Get the orchestrator history from the tracker API config. This is only 100% accurate if used in the orchestrator; otherwise, it may be ambiguous because the orchestrator is executed in parallel with story execution.

Returns:

• list[str] - The orchestrator history.

livechat_handover​

def livechat_handover(group_name: str | None = None) -> Any

Initiate a handover to a live agent. This method is used to initiate a handover to a live agent. The handover is initiated as soon as this method is called.

Arguments:

• group_name - Optional[str]: Name of the destination group that this conversation should be handed over to.

FlowManipulationApi Objects​

class FlowManipulationApi(FlowApi)

API for full flow manipulation including story transitions.

Extends FlowApi with the ability to force story transitions and control flow execution. Only available in RemoteLLMTrackerApi (in-story execution).

get_flows​

def get_flows() -> list[BaseStoryDefinition]

Get all story/flow definitions.

Returns:

• list[BaseStoryDefinition] - All story definitions (including substories, disabled, and error stories).

force_next_flow​

def force_next_flow(next_story_name: str | None = None,
                    switch_after_node: bool = False)

Force a flow/story transition.

switch_after_node controls timing:

• False: Overrides the next orchestrator decision when it is reached, so the next orchestrator-selected flow becomes next_story_name.
• True: Finishes only the current node, then jumps directly to next_story_name even if no orchestrator decision would be due yet.

This call is ignored when a tracker reset has already been triggered in the same execution cycle.

Arguments:

• next_story_name str | None, optional - Destination story name. If None, uses the active story. Default is None.
• switch_after_node bool, optional - Controls whether the flow change waits for the next orchestrator decision or happens immediately after the current node completes. Default is False.

shared.tracker.apis._llm_call_api

LLMCallApiConfig Objects​

class LLMCallApiConfig(FunctionExecutionApiConfig, TextNormalizeApiConfig)

Configuration for LLM call APIs with provider settings.

LLMCallApi Objects​

class LLMCallApi(FunctionExecutionApi, TextNormalizeApi)

API for making LLM calls including tool calls, structured output, and embeddings.

Provides comprehensive methods for LLM interactions, document retrieval, and semantic search operations.

get_enabled_chat_providers​

def get_enabled_chat_providers() -> dict[str, RemoteChatProviderConfig]

Get the enabled providers for the current LLM API Config.

Code Example:

    # Get all providers that are marked as enabled
    all_enabled_provider = tracker_api.get_enabled_chat_providers()

    # Use an alternative provider for specific subtasks within this action (use the configured provider name as key here)
    alternative_provider = all_enabled_provider["custom-openai"]

    api_response: StreamingChatApiRequest = await tracker_api.stream_call_llm(
        messages=[
        {"role": "system", "content": "You are a funny bard in a medieval court"},
        {"role": "user", "content": tracker_api.get_latest_user_message().content}
        ],
        # pass the provider to the llm call
        alternative_provider=alternative_provider
    )
    await tracker_api.stream_utter(api_response)

Returns:

• list[RemoteChatProviderConfig] - list of enabled providers

get_enabled_providers​

def get_enabled_providers() -> dict[str, RemoteChatProviderConfig]

Get the enabled providers for the current LLM API Config.

Code Example:

    # Get all providers that are marked as enabled
    all_enabled_provider = tracker_api.get_enabled_providers()

    # Use an alternative provider for specific subtasks within this action (use the configured provider name as key here)
    alternative_provider = all_enabled_provider["custom-openai"]

    api_response: StreamingChatApiRequest = await tracker_api.stream_call_llm(
        messages=[
        {"role": "system", "content": "You are a funny bard in a medieval court"},
        {"role": "user", "content": tracker_api.get_latest_user_message().content}
        ],
        # pass the provider to the llm call
        alternative_provider=alternative_provider
    )
    await tracker_api.stream_utter(api_response)

Returns:

• list[RemoteProviderConfig] - list of enabled providers

a_infer_with_chat_model​

@registered_llm_call
async def a_infer_with_chat_model(
        system_prompt: str,
        history: list[dict[str, Any]] | list[ConversationTurnType]
    | None = None,
        alternative_provider: RemoteChatProviderConfig | None = None,
        use_normalize: bool | None = None,
        **kwargs: Any) -> ChatApiResponse

Use a Chat Model to infer a response to a given system prompt. This method is used to generate a response to a given system prompt. The history parameter can be used to provide additional context to the model such as the recent conversation history.

Code Example:

response: ChatApiResponse = await tracker_api.a_infer_with_chat_model(
    system_prompt="You are a medieval bard",
    history=tracker_api.get_history(4)
)

    tracker_api.llm_utter(response)
    ```

**Arguments**:

- `system_prompt` _str_ - The system prompt to generate a response
- `history` _List[Dict] | List[UserTurn | AssistantTurn | ToolTurn | SystemTurn], optional_ - Optionally a list of ConversationTurns
  that are used as the conversation history by the called Model. Defaults to [].
- `alternative_provider` _Optional[RemoteProviderConfig]_ - An alternative configured provider to use for this call. Get all enabled
  providers via `tracker_api.get_enabled_providers()`
- `use_normalize` _Optional[bool]_ - Override text normalization for this call. If True, forces normalization. If False, disables it.
  If None, uses default settings.
  

**Returns**:

- `ChatApiResponse` - The response from the LLM API

<a id="shared.tracker.apis._llm_call_api.LLMCallApi.a_call_llm_w_tools"></a>

#### a\_call\_llm\_w\_tools

```python
@registered_llm_call
async def a_call_llm_w_tools(
        messages: list[dict[str, Any]] | list[ConversationTurnType],
        tools: list[dict],
        tool_choice: ChatCompletionToolChoiceOptionParam = "none",
        alternative_provider: RemoteChatProviderConfig | None = None,
        max_retries: int = 3,
        timeout: int | None = None,
        use_normalize: bool | None = None,
        **kwargs: Any) -> ChatApiResponse

Call the LLM API with a list of messages and tools. This method is used to call the LLM API with a list of messages and tools. The tool_choice parameter can be used to specify which tool should be used by the model. Tools are used to provide additional context to the model, e.g. the model can be asked to use a specific tool to generate a response or to extract certain inputs from the conversation history.

Code Example:

    tools = [
        {
            "type": "function",
            "function": {
                "name": "get_delivery_date",
                "description": "Get the delivery date for a customer's order. Call this whenever you need to know the delivery date,
                for example when a customer asks 'Where is my package'",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "order_id": {
                            "type": "string",
                            "description": "The customer's order ID.",
                        },
                    },
                    "required": ["order_id"],
                    "additionalProperties": False,
                },
            }
        }
        ]

        messages = [
            {
                "role": "system",
                "content": "You are a helpful customer support assistant. Use the supplied tools to assist the user."
            },
            {
                "role": "user",
                "content": "Hi, can you tell me the delivery date for my order?",
            },
        ]

        response: ChatApiResponse = await tracker_api.a_call_llm_w_tools(
        messages=messages,
        tools=tools
        )
        tool_calls = response.completion.choices[0].message.tool_calls

Arguments:

• messages _ List[Dict] | List[UserTurn | AssistantTurn | ToolTurn | SystemTurn]_ - The conversation history
• tools List - The tools that the LLM can use. Must conform to the OpenAI Tool API
• tool_choice ChatCompletionToolChoiceOptionParam, optional - Either "none" or "auto" Must conform the OpenAI Tool API. Defaults to "none".
• alternative_provider Optional[RemoteProviderConfig] - An alternative configured provider to use for this call. Get all enabled providers via tracker_api.get_enabled_providers()
• max_retries int, optional - The maximum number of retries for the LLM API call. Defaults to 3.
• timeout int, optional - The timeout for the LLM API call in seconds. Defaults to LLM_DEFAULT_TIMEOUT_SECONDS.
• use_normalize Optional[bool] - Override text normalization for this call. If True, forces normalization. If False, disables it. If None, uses default settings.

Returns:

• ChatApiResponse - The response from the LLM API

a_call_llm​

@registered_llm_call
async def a_call_llm(messages: list[dict[str, Any]]
                     | list[ConversationTurnType],
                     enforce_json: bool = False,
                     alternative_provider: RemoteChatProviderConfig
                     | None = None,
                     max_retries: int = 3,
                     timeout: int | None = None,
                     use_normalize: bool | None = None,
                     **kwargs: Any) -> ChatApiResponse

Call an LLM with a list of messages. This method is used to call the LLM API with a list of messages. The enforce_json parameter can be used to enforce the JSON format of the LLM Output.

IF you set enforce_json to True, the LLM API will return the raw JSON output of the model. This is useful if you want to use the raw output of the model in your application. But be aware that you should adapt your prompt accordingly so the model knows that it should return a JSON object.

Arguments:

• messages List[Dict] | List[UserTurn | AssistantTurn | ToolTurn | SystemTurn] - The conversation history or any other messages that should be used as input for the model
• enforce_json bool, optional - Set this to force the model to return a JSON Object in the response. Adapt your prompt to tell the model to return JSON!. Defaults to False.
• alternative_provider Optional[RemoteProviderConfig] - An alternative configured provider to use for this call. Get all enabled providers via tracker_api.get_enabled_providers()
• max_retries int, optional - The maximum number of retries for the LLM API call. Defaults to 3.
• timeout int, optional - The timeout for the LLM API call in seconds. Defaults to LLM_DEFAULT_TIMEOUT_SECONDS.
• use_normalize Optional[bool] - Override text normalization for this call. If True, forces normalization. If False, disables it. If None, uses default settings.

Returns:

• ChatApiResponse - The response from the LLM API

stream_call_llm​

async def stream_call_llm(
        messages: list[dict[str, Any]] | list[ConversationTurnType],
        enforce_json: bool = False,
        alternative_provider: RemoteChatProviderConfig | None = None,
        tools: list[dict] | None = None,
        tool_choice: ChatCompletionToolChoiceOptionParam = "none",
        max_retries: int = 3,
        timeout: int | None = None,
        use_normalize: bool | None = None,
        **kwargs: Any) -> StreamingChatApiRequest

Call an LLM with a list of messages. This method is used to call the LLM API with a list of messages. The enforce_json parameter can be used to enforce the JSON format of the LLM Output.

IF you set enforce_json to True, the LLM API will return the raw JSON output of the model. This is useful if you want to use the raw output of the model in your application. But be aware that you should adapt your prompt accordingly so the model knows that it should return a JSON object.

Arguments:

• messages List[Dict] | List[UserTurn | AssistantTurn | ToolTurn | SystemTurn] - The conversation history or any other messages that should be used as input for the model
• enforce_json bool, optional - Set this to force the model to return a JSON Object in the response. Adapt your prompt to tell the model to return JSON!. Defaults to False.
• alternative_provider Optional[RemoteProviderConfig] - An alternative configured provider to use for this call. Get all enabled providers via tracker_api.get_enabled_providers()
• max_retries int, optional - The maximum number of retries for the LLM API call. Defaults to 3.
• timeout int, optional - The timeout for the LLM API call in seconds. Defaults to LLM_DEFAULT_TIMEOUT_SECONDS.
• use_normalize Optional[bool] - Override text normalization for this call. If True, forces normalization. If False, disables it. If None, uses default settings.

Returns:

• StreamingChatApiRequest - The response from the LLM API

a_stream_call_llm_w_tools​

async def a_stream_call_llm_w_tools(
        messages: list[dict[str, Any]] | list[ConversationTurnType],
        tools: list[dict],
        tool_choice: ChatCompletionToolChoiceOptionParam = "auto",
        alternative_provider: RemoteChatProviderConfig | None = None,
        num_chunks_to_inspect: int = 3,
        mixed_streaming: bool = False,
        use_normalize: bool | None = None,
        **kwargs: Any) -> ClassifyingStreamingChatApiRequest

Call the LLM API with a list of messages and tools, returning a streaming response. This method streams the LLM's output, supporting both regular chat responses and tool calls. When mixed_streaming=False (default), the first part of the stream is inspected to determine the response type: "text_response", "tool_call", or (when mixed_streaming=True) "text_and_tool_call" for mixed streams.

Arguments:

• messages List[Dict] | List[UserTurn | AssistantTurn | ToolTurn | SystemTurn] - The conversation history.
• tools List[Dict] - The tools available to the LLM, conforming to the OpenAI Tool API.
• tool_choice ChatCompletionToolChoiceOptionParam, optional - Tool selection mode ("none", "auto", or a specific tool). Defaults to "auto".
• alternative_provider Optional[RemoteProviderConfig] - Optionally use a different provider for this call.
• num_chunks_to_inspect int, optional - Number of initial stream chunks to inspect for response classification. Defaults to 3.
• mixed_streaming bool, optional - If False, classify stream as "text_response" or "tool_call". If True, stream_type is "text_and_tool_call".
• use_normalize Optional[bool] - Override text normalization for this call. If True, forces normalization. If False, disables it. If None, uses default settings.

Returns:

• ClassifyingStreamingChatApiRequest - The streaming response object, which can yield either chat tokens or tool call events.

a_call_llm_with_response_model​

@registered_llm_call
async def a_call_llm_with_response_model(
        messages: list[dict[str, Any]] | list[ConversationTurnType],
        response_model: type[DynamicExtractorModel] | type[BaseModel],
        alternative_provider: RemoteChatProviderConfig | None = None,
        max_retries: int = 3,
        timeout: int | None = None,
        use_normalize: bool | None = None,
        **kwargs: Any) -> ChatApiResponse

** Expert Usage only - Needs to be supported by used Provider ** Call an LLM with a list of messages and a response model. This method is used to call the LLM API with a list of messages and a response model. The response model is used to guide the output of the model.

Arguments:

• messages List[Dict] | List[UserTurn | AssistantTurn | ToolTurn | SystemTurn] - The conversation history or any other messages that should be used as input for the model
• response_model type[DynamicExtractorModel] - A Pydantic Model that should be used to guide the output of the model -- needs to be supported by the Provider API e.g. custom providers
• alternative_provider Optional[RemoteProviderConfig] - An alternative configured provider to use for this call. Get all enabled providers via tracker_api.get_enabled_providers()
• max_retries int, optional - The maximum number of retries for the LLM API call. Defaults to 3.
• timeout int, optional - The timeout for the LLM API call in seconds. Defaults to LLM_DEFAULT_TIMEOUT_SECONDS.
• use_normalize Optional[bool] - Override text normalization for this call. If True, forces normalization. If False, disables it. If None, uses default settings.

Returns:

• ChatApiResponse - The response from the LLM API

a_stream_call_response_model​

async def a_stream_call_response_model(
        messages: list[dict[str, Any]] | list[ConversationTurnType],
        response_model: type[DynamicExtractorModel] | type[BaseModel],
        alternative_provider: RemoteChatProviderConfig | None = None,
        max_retries: int = 3,
        timeout: int | None = None,
        use_normalize: bool | None = None,
        **kwargs: Any) -> StreamingChatApiRequest

Call an LLM with a list of messages and a response model in streaming mode. Use this together with stream_response_model to stream only the value of a specific key from the structured JSON output to the user.

Code Example:

stream_chat_api_response = await tracker_api.a_stream_call_response_model(
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": tracker_api.get_latest_user_message().content},
    ],
    response_model=MyResponseModel,
)
streamed = await tracker_api.stream_response_model(
    stream_chat_api_response,
    stream_key="answer",
)

Arguments:

• messages List[Dict] | List[UserTurn | AssistantTurn | ToolTurn | SystemTurn] - The conversation history or any other messages that should be used as input for the model response_model (type[DynamicExtractorModel] | type[BaseModel]): A Pydantic Model that should be used to guide the output of the model.
• alternative_provider Optional[RemoteProviderConfig] - An alternative configured provider to use for this call.
• max_retries int, optional - The maximum number of retries for the LLM API call. Defaults to 3.
• timeout int, optional - The timeout for the LLM API call in seconds. Defaults to LLM_DEFAULT_TIMEOUT_SECONDS.
• use_normalize Optional[bool] - Override text normalization for this call.

Returns:

• StreamingChatApiRequest - The streaming response from the LLM API

a_create_embedding​

async def a_create_embedding(input: str,
                             provider_title: str | None = None,
                             max_retries: int = 3,
                             timeout: int | None = None,
                             **kwargs: Any) -> CreateEmbeddingResponse

Create an embedding for a given input. This method is used to create an embedding for a given input. The input can be any text that should be embedded by the embedding_model of the currently configured Provider.

Arguments:

• input str - The text to embed
• provider_title Optional[str], optional - The title of the provider to use for the embedding. If None is passed the default provider will be used. Defaults to None.
• max_retries int, optional - The maximum number of retries for the embedding creation. Defaults to 3.
• timeout int, optional - The timeout for the embedding creation in seconds. Defaults to LLM_DEFAULT_TIMEOUT_SECONDS.

Returns:

• CreateEmbeddingResponse - The response from the LLM API

a_points_query​

async def a_points_query(request: dict[str, Any]) -> dict[str, Any]

Perform a /points/query request against the configured VectorDB. This method is used to query the VectorDB with a given request. The request must conform to the VectorDB API.

Arguments:

checkout https://qdrant.tech/documentation/concepts/hybrid-queries/

a_points_scroll​

async def a_points_scroll(request: dict[str, Any]) -> dict[str, Any]

Perform a /points/scroll request against the configured VectorDB. This method is used to scroll the VectorDB with a given request. The request must conform to the VectorDB API.

Arguments:

checkout https://api.qdrant.tech/v-1-13-x/api-reference/points/scroll-points

a_hybrid_search​

async def a_hybrid_search(
        query: str,
        prefetch_requests: list[PrefetchRequest],
        hybrid_search_params: HybridSearchParams | None = None,
        max_retries: int = 3,
        timeout: int | None = None) -> RemoteSemanticSearchResponse

This method executes a hybrid search combining both sparse (BM25) and dense vector search capabilities against the VectorDB. It processes multiple prefetch requests in sequence, handling both sparse and dense embeddings appropriately.

Arguments:

• query str - The search query text to be used for the search
• prefetch_requests List[PrefetchRequest] - List of prefetch request objects that specify search parameters for different providers. Each request can include:
  • provider: Provider name ("Qdrant/bm25" for sparse or custom provider for dense)
  • filter: Dictionary of filter conditions
  • tags: List of tags to filter results
  • query: Optional specific query text (defaults to main query if not provided)
• hybrid_search_params HybridSearchParams, optional - Parameters for hybrid search configuration. Defaults to default HybridSearchParams instance.
• max_retries int, optional - The maximum number of retries for the LLM API call. Defaults to 3.
• timeout int, optional - The timeout for the LLM API call in seconds. Defaults to LLM_DEFAULT_TIMEOUT_SECONDS.

Returns:

• dict - JSON response from the vector database containing search results

Raises:

• Exception - If the VectorDB service request fails with status code >= 300

Example:

    res = await tracker_api.a_hybrid_search(
        query="valar",
        prefetch_requests=[
            PrefetchRequest(
                provider="Qdrant/bm25",
                filter={"must": [{"key": "metadata.filename", "match": {"value": "valar dohaeris"}}]},
            ),
            PrefetchRequest(provider="Botario__Default", tags=["Game Of Thrones"]),
        ],
        hybrid_search_params=HybridSearchParams(score_threshold=0.1),
    )

a_semantic_search​

async def a_semantic_search(
        query: str,
        filters: list[RagMetaDataFilter] | dict[str, Any] | None = None,
        top_k: int = 5,
        score_threshold: float = 0.5,
        tags: list[str] | None = None,
        providers: Literal["all"] | list[str] | None = None,
        max_retries: int = 3,
        timeout: int | None = None) -> RemoteSemanticSearchResponse

Perform semantic search for a given query against the configured Knowledge in the VectorDB.

Arguments:

• query str - The Search string
• filters Optional[List[RagMetaDataFilter] | Dict[str, Any]], optional - A list of custom filters to use e.g. metadata filters that are set for chunks in the VectorDB. If a dict is passed it will be interpreted as exact match filters.Defaults to None.
• top_k Optional[int], optional - How many chunks should be returned at max. Defaults to 5.
• score_threshold Optional[float], optional - Whats the minimus similarity score a chunk needs to meet. Defaults to None.
• tags Optional[List[str]], optional - Use tags to filter tagged chunks. Defaults to None.
• providers Optional[Literal["all"] | List[str]], optional - Use this to specify which providers should be used for the search. Defaults to None and uses the configured default knowledge provider.
• max_retries int, optional - The maximum number of retries for the LLM API call. Defaults to 3.
• timeout int, optional - The timeout for the LLM API call in seconds. Defaults to LLM_DEFAULT_TIMEOUT_SECONDS.

Returns:

• RemoteSemanticSearchResponse - Returns the list of Chunks that were found

validate_with_BaseModel​

async def validate_with_BaseModel(
        extracted_json: dict[str, Any],
        DynamicModel: type[DynamicExtractorModel]) -> DynamicExtractorModel

Helper Method to validate the extracted JSON with a Pydantic BaseModel. Calls the model_validate method of the DynamicModel which will execute all defined validation methods for the corresponding ExtractionFields of this FormNode.

Arguments:

• extracted_json dict[str, Any] - The raw json that was extracted from the user input
• DynamicModel type[DynamicExtractorModel] - The BaseModel that should be used for Validation. In most cases this will be a DynamicModel built from all ExtractionFields of the the FormNode of the current FlowStep.

Returns:

• DynamicExtractorModel - On success the validated DynamicModel is returned. If the validation fails an Exception is raised.

shared.tracker.apis._authorized_api

AuthorizedApiConfig Objects​

class AuthorizedApiConfig(BaseApiConfig)

Configuration for APIs requiring authentication.

AuthorizedApi Objects​

class AuthorizedApi(BaseApi)

API with authentication support for secure communication.

Extends BaseApi with authentication token handling for secure communication with LLM service.

shared.tracker.apis._text_normalize_api

TextNormalizeApiConfig Objects​

class TextNormalizeApiConfig(BaseApiConfig)

Configuration for text normalization APIs.

TextNormalizeApi Objects​

class TextNormalizeApi(BaseApi)

API for text normalization capabilities.

Provides methods to check and apply text normalization settings at system, node, and function levels.

shared.tracker.apis.tag_api

shared.tracker.apis._slot_api

SlotApiConfig Objects​

class SlotApiConfig(StateManipulationApiConfig)

Configuration for slot API containing current slot values.

SlotApi Objects​

class SlotApi(StateManipulationApi)

API for reading slot (ExtractionField) values.

Provides methods to access slot values with optional state synchronization. Read-only access to slots.

slots​

@property
@StateManipulationApi.guarded_state
def slots() -> dict[str, Any]

Get the slots of the tracker. This method returns the slots of the tracker. The slots are returned as a dictionary of slot names and values. .. deprecated:: This method is deprecated and will return unsynchronized state values. To make sure you get the latest state values, please use the async method a_slots.

Returns:

dict[str, Any]: The slots of the tracker.

a_slots​

@property
async def a_slots() -> dict[str, Any]

Get the slots of the tracker. This method returns the slots of the tracker. The slots are returned as a dictionary of slot names and values.

Returns:

dict[str, Any]: The slots of the tracker.

a_get_slot_value​

async def a_get_slot_value(slot_name: str,
                           default: Any | None = None,
                           include_confidential: bool = False) -> Any

Get the value of a slot (ExtractionField). Slot or ExtractionField values are set either by Agents in FormNodes or directly via the tracker_api in custom scripts. Synchronizes the state from the backend before returning the slot value.

Arguments:

• slot_name str - Name of the slot (ExtractionField)

• default - Default value to return if the slot is not set.

• include_confidential bool - If True and a confidential slot with this name was injected for the current turn, return its plaintext value instead of the masked state value. Confidential slots are request-scoped and never persisted.

• WARNING - The returned value is sensitive. Once retrieved, the caller is responsible for ensuring it is not explicitly logged, stored, or otherwise exposed.

Returns:

• Any - Value of the slot or the given default value

get_slot_value​

def get_slot_value(slot_name: str,
                   default: Any | None = None,
                   include_confidential: bool = False) -> Any

Get the value of a slot (ExtractionField). Slot or ExtractionField values are set either by Agents in FormNodes or directly via the tracker_api in custom scripts.

.. deprecated:: This method is deprecated and will return unsynchronized state values. To make sure you get the latest state values, please use the async method a_get_slot_value.

Arguments:

• slot_name str - Name of the slot (ExtractionField)

• default - Default value to return if the slot is not set.

• include_confidential bool - If True and a confidential slot with this name was injected for the current turn, return its plaintext value instead of the masked state value. Confidential slots are request-scoped and never persisted.

• WARNING - The returned value is sensitive. Once retrieved, the caller is responsible for ensuring it is not explicitly logged, stored, or otherwise exposed.

Returns:

• Any - Value of the slot or the given default value

SlotManipulationApi Objects​

class SlotManipulationApi(SlotApi)

API for setting and manipulating slot values.

Extends SlotApi with write capabilities to set slot values. Only available in RemoteLLMTrackerApi (in-story execution).

set_slot​

def set_slot(name: str, value: Any)

Set a slot (ExtractionField) value in the tracker. Slots (ExtractionField) are used to store information about the conversation, e.g. user name, age etc. Slots are set by the backend as soon as this event is processed.

Arguments:

• name str - Name of the slot - must not contain special characters or blanks
• value Any - Value of the slot

shared.tracker.apis._base

LoggingConfig Objects​

class LoggingConfig(BaseModel)

Configuration for logging with bot and session identifiers.

BaseApiConfig Objects​

class BaseApiConfig(BaseModel)

Base configuration for all tracker APIs with bot and session context.

BaseApi Objects​

class BaseApi()

Base API providing fundamental capabilities for all tracker APIs.

Provides logging, event buffering, test mode detection, and URL access. Foundation for all other tracker API classes.

get_custom_token_payload​

def get_custom_token_payload() -> dict[str, Any] | None

Get the custom token payload data (e.g., EntraID claims etc.).

The payload is request-scoped and never persisted.

WARNING: The returned value is sensitive. Once retrieved, the caller is responsible for ensuring it is not logged, stored, or otherwise exposed. The platform guarantees protection only up to this point.

Returns:

The token payload as a dict, or None if not set.

log​

def log(payload: str | dict,
        level: Literal["DEBUG", "INFO", "WARNING", "ERROR",
                       "CRITICAL"] = "INFO")

Log a message or object and persist the log in the tracker.

Arguments:

• payload Union[str, dict] - The message to log can either be a string or a dictionary.

debug​

def debug(payload: str | dict)

Log a debug message or object and persist the log in the tracker.

Arguments:

• payload str | dict - The message to log can either be a string or a dictionary.

info​

def info(payload: str | dict)

Log an info message or object and persist the log in the tracker.

Arguments:

• payload str | dict - The message to log can either be a string or a dictionary.

warning​

def warning(payload: str | dict)

Log a warning message or object and persist the log in the tracker.

Arguments:

• payload str | dict - The message to log can either be a string or a dictionary.

error​

def error(payload: str | dict)

Log an error message or object and persist the log in the tracker.

Arguments:

• payload str | dict - The message to log can either be a string or a dictionary.

critical​

def critical(payload: str | dict)

Log a critical message or object and persist the log in the tracker.

Arguments:

• payload str | dict - The message to log can either be a string or a dictionary.

get_start_url​

def get_start_url() -> str | None

Get the start URL for the current bot. The start URL stores the URL from which the user's first interaction originates.

Returns:

• str - The start URL for the current bot. Can be None if the Chat was started via API and the Client did not provide a start URL in the ChatRequest.

get_current_url​

def get_current_url() -> str | None

Get the current URL for the current bot. The start URL stores the URL from which the user's first interaction originates.

Returns:

• str - The start URL for the current bot. Can be None if the Chat was started via API and the Client did not provide a start URL in the ChatRequest.

get_user​

def get_user() -> User | None

Get the user object for the current bot. The user contains information about who is interacting with the bot.

Returns:

User | None: The user object for the current bot. Can be None if the Chat was started via API and the Client did not provide a user in the ChatRequest. Will be None if identified chatting in the knowledge manager settings is disabled.

shared.tracker.apis.function_execution_api

shared.tracker.apis._legacy_apis

PreStoryAgentTrackerApiConfig Objects​

class PreStoryAgentTrackerApiConfig(BaseAssistantApiConfig, SlotApiConfig,
                                    TagApiConfig, PreStoryFlowApiConfig,
                                    UtterApiConfig, LLMCallApiConfig,
                                    PhoneApiConfig)

Configuration for tracker APIs used by pre-story agents.

This configuration class defines the settings and context needed for agents that execute BEFORE a story is selected or started (orchestrator, guard, preprocessing agents).

Execution Context:

• Used during the pre-story phase (before story selection/execution)
• Provides read-only access to conversation state
• Allows emitting events
• Does NOT allow flow/story manipulation

Composition: Inherits capabilities from multiple config classes:

• BaseAssistantApiConfig: Core bot/session/tenant configuration
• SlotApiConfig: Slot access and state configuration
• PreStoryFlowApiConfig: Limited flow information (no manipulation)
• UtterApiConfig: Response generation and postprocessing configuration
• LLMCallApiConfig: LLM provider configuration
• PhoneApiConfig: Voice/phone interaction configuration

Attributes:

• execution_mode - Set to "pre_story" indicating pre-story execution phase

Notes:

This config is used by PreFlowRemoteLLMTrackerApi which has restricted capabilities compared to RemoteLLMTrackerApi (used in-story).

TrackerApiConfig Objects​

class TrackerApiConfig(PreStoryAgentTrackerApiConfig, FlowApiConfig)

Configuration for tracker APIs used by in-story agents.

This configuration extends PreStoryAgentTrackerApiConfig with additional capabilities needed for agents that execute DURING story execution (form nodes, response nodes, RAG nodes).

Execution Context:

• Used during the in-story phase (after story selection, during story execution)
• Provides full read/write access to conversation and flow state
• Allows flow manipulation (story transitions, node navigation)
• Allows story state manipulation

Composition: Inherits all capabilities from PreStoryAgentTrackerApiConfig PLUS:

• FlowApiConfig: Full flow manipulation capabilities (story transitions, etc.)

Attributes:

• execution_mode - Set to "in_story" indicating in-story execution phase

Notes:

This config is used by RemoteLLMTrackerApi which has full capabilities for manipulating conversation flow during story execution.

TrackerApi Objects​

class TrackerApi(BaseAssistantApi, FlowApi, PhoneApi)

Base tracker API providing core conversation state access.

TrackerApi is the foundational class for all tracker APIs, providing basic access to conversation state, flow information, and phone interaction capabilities. It does NOT include LLM calling capabilities or slot manipulation.

Core Capabilities:

• Read access to conversation state (slots, history, active story)
• Basic flow information access
• Phone/voice interaction methods
• Event buffering for agent outputs

Attributes:

• tracker_api_config - Configuration object containing conversation context

• botId - Identifier of the current bot

• tenantId - Identifier of the tenant/organization

• sessionId - Unique identifier for the conversation session

• _slots - Current slot values (protected, use SlotApi for manipulation)

• latest_user_msg - Most recent user message

• active_story - Currently active story name

• event_buffer - Queue for collecting events emitted by the agent

• _history - Conversation history (protected)

• action_service_manager_config - Configuration for ActionsManager communication

  Inheritance Chain: TrackerApi → PreFlowLLMTrackerApi → PreFlowRemoteLLMTrackerApi → RemoteLLMTrackerApi

Notes:

This class is typically not used directly. Use PreFlowRemoteLLMTrackerApi or RemoteLLMTrackerApi which extend this with LLM and slot manipulation capabilities.

PreFlowLLMTrackerApi Objects​

class PreFlowLLMTrackerApi(TrackerApi, UtterApi, LLMBaseApi, LLMCallApi,
                           TagManipulationApi)

Tracker API with LLM calling and utterance capabilities.

PreFlowLLMTrackerApi extends TrackerApi with the ability to make LLM API calls and generate utterances. It provides the foundation for agents that need to interact with LLMs but still lacks slot manipulation and full flow control.

Additional Capabilities (beyond TrackerApi):

• LLM API calls (a_call_llm, a_call_llm_with_response_model, etc.)
• Response generation (utter_llm, stream_llm_response, fixed_utter)
• Conversation history management
• Post-processing of LLM responses

Composition: Inherits from:

• TrackerApi: Core conversation state access
• UtterApi: Response generation and emission
• LLMBaseApi: Base LLM interaction capabilities
• LLMCallApi: Advanced LLM calling methods (tool calls, structured output)

Use Cases: This is an intermediate class in the hierarchy. Not typically used directly. Extended by PreFlowRemoteLLMTrackerApi which adds slot manipulation.

Inheritance Chain: TrackerApi → PreFlowLLMTrackerApi → PreFlowRemoteLLMTrackerApi → RemoteLLMTrackerApi

Notes:

This class provides LLM capabilities but does NOT provide slot manipulation (that comes from SlotApi/SlotManipulationApi in subclasses).

PreFlowRemoteLLMTrackerApi Objects​

class PreFlowRemoteLLMTrackerApi(PreFlowLLMTrackerApi, SlotApi, FlowApi,
                                 PhoneApi)

Tracker API for pre-story agents with read-only flow access.

PreFlowRemoteLLMTrackerApi is used by agents that execute BEFORE a story is selected or started (orchestrator, guard, preprocessing agents). It provides full LLM capabilities and slot manipulation, but with LIMITED flow control - agents can read flow state but cannot manipulate it (no story transitions, node changes, etc.).

When to Use:

• Orchestrator agents (story routing/selection)
• Guard agents (safety validation before story execution)
• Preprocessing agents (input transformation before story execution)
• Mail preprocessing agents (email content preprocessing)

Execution Phase: PRE-STORY: Executes before a story is selected and started

• Cannot change active story
• Cannot trigger story transitions
• Cannot manipulate flow state
• Can emit events that are applied before story execution

Full Capabilities: ✓ Read conversation state (slots, history, active story) ✓ Make LLM API calls (structured output, tool calling, streaming) ✓ Read slots ✓ Generate responses (utter_llm, fixed_utter, stream responses) ✓ Read flow information ✓ Phone/voice interactions ✗ Manipulate flow/story state (NO transitions, NO node changes) ✗ Access advanced flow manipulation (use RemoteLLMTrackerApi for this) ✗ Set slots

Composition: Inherits from:

• PreFlowLLMTrackerApi: LLM calling and utterance capabilities
• SlotApi: Slot reading and manipulation (set_slot, update_slots, get_slot_value)
• FlowApi: Flow information access (read-only in pre-story context)
• PhoneApi: Voice/phone interaction capabilities

Key Distinction from RemoteLLMTrackerApi:

• PreFlowRemoteLLMTrackerApi: Used BEFORE story execution (limited flow control)
• RemoteLLMTrackerApi: Used DURING story execution (full flow manipulation)

Example:

In an orchestrator agent

async def predict_story(self, tracker_api: PreFlowRemoteLLMTrackerApi, ...): ... # Can access slots ... user_language = tracker_api.get_slot_value("language") ... # Can make LLM calls ... response = await tracker_api.a_call_llm_w_tools(...) ... # CANNOT set slots ... tracker_api.set_slot("detected_intent", "booking") # ✗ Not available ... # CANNOT manipulate flow (no story transitions) ... tracker_api.force_next_flow(...) # ✗ Not available

Inheritance Chain: TrackerApi → PreFlowLLMTrackerApi → PreFlowRemoteLLMTrackerApi → RemoteLLMTrackerApi

RemoteLLMTrackerApi Objects​

class RemoteLLMTrackerApi(PreFlowRemoteLLMTrackerApi, FlowManipulationApi,
                          SlotManipulationApi)

Full-featured tracker API for in-story agents with complete flow manipulation.

RemoteLLMTrackerApi is the most capable tracker API, used by agents that execute DURING story execution (form nodes, response nodes, RAG nodes). It provides ALL capabilities including full flow manipulation, story transitions, and advanced slot operations.

When to Use:

• Form node agents (extraction agents in FormNodes)
• Response node agents (dynamic response generation in ResponseNodes)
• RAG node agents (knowledge retrieval and generation in RAGNodes)
• Custom action execution
• Any agent executing within an active story

Execution Phase: IN-STORY: Executes during story execution (after story selection)

• Can manipulate active story state
• Can trigger story transitions
• Can change flow execution
• Full read/write access to all conversation state
• Can execute custom actions
• Can perform advanced slot operations

Full Capabilities: ✓ All capabilities from PreFlowRemoteLLMTrackerApi PLUS: ✓ Flow manipulation (trigger_story_transition, set_story, etc.) ✓ Advanced slot operations (delete_slots, clear_slots, batch operations) ✓ Execute custom actions ✓ Knowledge base retrieval (retrieve_documents, retrieve_with_reranking) ✓ Full story lifecycle control ✓ Node navigation and flow control

Composition: Inherits from:

• PreFlowRemoteLLMTrackerApi: Pre-story capabilities (LLM, slots, utterances)
• FlowManipulationApi: FULL flow control (story transitions, flow changes)
• SlotManipulationApi: ADVANCED slot operations (delete, clear, batch ops)

Additional Parameters: llm_api_config: Configuration for LLM provider and parameters actions: Dictionary of available custom actions auth_token: Authentication token for secure communication

Key Distinction from PreFlowRemoteLLMTrackerApi:

• PreFlowRemoteLLMTrackerApi: Limited flow control (read-only, pre-story phase)
• RemoteLLMTrackerApi: FULL flow manipulation (read/write, in-story phase)

Example:

In a form node agent

async def run_extractors(self, tracker_api: RemoteLLMTrackerApi, ...): ... # All PreFlowRemoteLLMTrackerApi capabilities ... user_input = tracker_api.get_slot_value("user_name") ... response = await tracker_api.a_call_llm_w_tools(...) ... tracker_api.set_slot("extracted_name", name) ... ... # PLUS full flow manipulation ... if special_condition: ... tracker_api.force_next_flow("special_story") ...

Attributes:

• type - Set to "remote" indicating remote execution context

  Inheritance Chain: TrackerApi → PreFlowLLMTrackerApi → PreFlowRemoteLLMTrackerApi → RemoteLLMTrackerApi

Notes:

This is the most commonly used tracker API in custom agents since most custom agents are executed within story nodes (form, response, RAG).

shared.tracker.apis._function_execution_api

FunctionExecutionApiConfig Objects​

class FunctionExecutionApiConfig(AuthorizedApiConfig)

Configuration for FunctionExecutionApi.

This config is used by APIs that need access to locally registered action functions.

FunctionExecutionApi Objects​

class FunctionExecutionApi(AuthorizedApi)

Base API class that provides access to locally registered action functions.

This class manages the actions dict which contains registered functions like:

• processor_{name}: Post-processing functions
• validator_{name}: Field validation functions
• form_validator_{node_name}: Form-level validation functions

Both UtterApi and LLMCallApi inherit from this class to access actions.

shared.tracker.apis.state_manipulation_api

shared.tracker.apis._conversation_snapshot_api

ConversationSnapshotApi Objects​

class ConversationSnapshotApi()

Read-only access to a frozen conversation state snapshot.

Provides conversation history, slot values, and message accessors without any live state synchronization.

Requires the inheriting class to set these instance attributes in its init: slots: dict[str, Any] history: list[ConversationTurnType] latest_user_msg: UserTurn | None _active_story_name: str | None

get_slot​

def get_slot(slot_name: str, default: Any | None = None) -> Any

Return the value of a slot from the conversation snapshot.

Arguments:

• slot_name - Name of the slot (ExtractionField).
• default - Value to return when the slot is not set.

Returns:

Deep copy of the slot value, or default if not present.

get_slots​

def get_slots() -> dict[str, Any]

Return all slots from the conversation snapshot.

get_history​

def get_history(n_turns: int | None = None,
                unsafe: bool = False) -> list[ConversationTurnType]

Return conversation history, optionally limited to the last n user turns.

Arguments:

• n_turns - Number of UserTurns to include (counting from the end). If None, returns all turns. If < 0, returns all. If 0, returns an empty list.
• unsafe - If True, restores original (pre-guard) content of flagged user turns.

get_latest_bot_message​

def get_latest_bot_message() -> AssistantTurn | None

Return the last assistant turn with non-empty content from the conversation history.

Skips function-call turns (AssistantFunctionCall) where content is None.

get_latest_user_message​

def get_latest_user_message() -> UserTurn

Return the latest user message from the conversation snapshot.

Raises:

• AssertionError - If no user message is present in the snapshot.

get_active_story_name​

def get_active_story_name() -> str | None

Return the name of the active story at the time of the snapshot.

shared.tracker.apis._phone_api

PhoneApiConfig Objects​

class PhoneApiConfig(BaseApiConfig)

Configuration for phone/voice APIs with STT/TTS provider settings.

PhoneApi Objects​

class PhoneApi(BaseApi)

API for phone and voice interaction capabilities.

Provides methods for call control (hangup, forward), STT/TTS configuration, recording, DTMF input, ambience sounds, and contact information management.

hangup​

def hangup()

Hangup the conversation. This method is used to hangup the conversation via Phone by yielding a HangupEvent to the client. The conversation is ended as soon as this method is called.

forward_call​

def forward_call(forward_call_id: str,
                 forward_call_ringing: int = 15,
                 sip_headers: dict[str, Any] | None = None)

Forward the call to another number. This method is used to forward the call to another number by yielding a ForwardEvent. The call is forwarded as soon as this method is called.

There are two modes for handling the result of a forwarded call:

1. Default Mode (Slot-Based): After the forwarded call ends, the Middleware will set the sys_phone_dialstatus slot with one of these values:

• "ANSWER" - The call was answered successfully
• "BUSY" - Called party was busy (includes DND mode with SIP 486 response)
• "NOANSWER" - Called party did not answer
• "CHANUNAVAIL" - Endpoint unreachable (not registered) or nonexistent location
• "CONGESTION" - Channel/switching congestion or slow/no response
• "CANCEL" - Call cancelled before being answered
• "DONTCALL" - Privacy mode: caller sent to 'Go Away' script
• "TORTURE" - Privacy mode: caller sent to 'torture' script
• "INVALIDARGS" - Invalid syntax in forwarding request

2. Legacy Mode (Signal-Based): After the call ends, the middleware triggers a flow based on these signal paths, you must define flows for the corresponding signals:

• "/ps_forward_answer" - Call answered successfully
• "/ps_forward_busy" - Called party busy/DND mode
• "/ps_forward_noanswer" - No answer received
• "/ps_forward_chanunavail" - Endpoint unreachable
• "/ps_forward_congestion" - Channel congestion
• "/ps_forward_cancel" - Call cancelled
• "/ps_forward_dontcall" - Privacy rejection
• "/ps_forward_torture" - Privacy rejection (torture)
• "/ps_forward_invalidargs" - Invalid syntax

For more details on dialstatus values, see Asterisk Dial documentation.

Arguments:

• forward_call_id str - The phone number to forward the call to
• forward_call_ringing int, optional - Number of seconds to ring before timeout. Defaults to 15.
• sip_headers dict[str, Any], optional - Additional SIP headers for the forwarded call. Defaults to {}.

update_stt_config​

def update_stt_config(update_payload: (BotarioSTTConfigUpdatePayload
                                       | AzureSTTConfigUpdatePayload
                                       | GoogleSTTConfigUpdatePayload),
                      provider_title: str | None = None)

Set the STT Configuration. This method is used to set the STT Configuration. The STT Configuration is set as soon as this method is called.

Arguments:

• update_payload BotarioSTTConfigUpdatePayload | AzureSTTConfigUpdatePayload | GoogleSTTConfigUpdatePayload - The update payload for the STT Configuration

update_tts_config​

def update_tts_config(update_payload: BotarioTTSConfigUpdatePayload
                      | AzureTTSConfigUpdatePayload
                      | GoogleTTSConfigUpdatePayload
                      | ElevenLabsTTSConfigUpdatePayload,
                      provider_title: str | None = None)

Set the TTS Configuration. This method is used to set the TTS Configuration. The TTS Configuration is set as soon as this method is called.

Arguments:

update_payload (BotarioTTSConfigUpdatePayload | AzureTTSConfigUpdatePayload | GoogleTTSConfigUpdatePayload |

• ElevenLabsTTSConfigUpdatePayload) - The update payload for the TTS Configuration
• provider_title Optional[str], optional - The title of the TTS provider to use. If not provided, uses the active TTS provider. Defaults to None.

enable_recording​

def enable_recording(enable: bool = True, include_current_turn: bool = False)

Enable or disable recording of the conversation. This method is used to enable or disable recording of the conversation. Recording is enabled or disabled as soon as this method is called.

Arguments:

• enable bool - True to enable recording, False to disable recording
• include_current_turn bool - If True, the current turn's audio will be saved immediately. If False (default), recording starts from the next user input.

disable_ambience_channel​

async def disable_ambience_channel(sound_name: Literal["tastatur",
                                                       "copy_machine",
                                                       "office",
                                                       "background_music"])

Disables the ambience channel on the conversation. sound_name: name of the ambience_sound that should be disabled

enable_ambience_channel​

async def enable_ambience_channel(sound_name: Literal["tastatur",
                                                      "copy_machine", "office",
                                                      "background_music"],
                                  volume: int = 0)

Enables the ambience channel on the conversation. sound_name: name of the ambience_sound volume: Positiv or negative integer, between -10 and 20

set_phone_channel_variables​

async def set_phone_channel_variables(variables: dict[str, Any])

Sets custom channel variables on the phone channel. variables: dictionary of channel variables to set

enable_dtmf​

async def enable_dtmf(termination_symbol: str = "#",
                      single_digit_mode: bool = False,
                      time_out: float = 15)

Enable DTMF input for the user. The user can now enter DTMF digits to interact with the bot. Only set values will be applied. Sensitive Defaults for the respective values are set in the third-party system (e.g. phoneserver).

Arguments:

• termination_symbol str - The symbol that the user must enter to indicate that their input is complete. Default is "#".
• single_digit_mode bool - If True, only a single digit is expected and no termination symbol is required. The input is automatically completed after one digit. Default is False.
• time_out Optional[float] - The time to wait for more digits after the last digit was received. Must not be negative. Default is 15.

shared.tracker.apis._config_map_api

ConfigMapApiConfig Objects​

class ConfigMapApiConfig(BaseApiConfig)

Configuration for APIs requiring config map access.

ConfigMapApi Objects​

class ConfigMapApi(BaseApi)

API providing configuration map access methods.

Provides methods to access bot configuration values and visual tags. Foundation for APIs that need to read configuration data.

get_visual_tag​

def get_visual_tag() -> str | None

Get the visual tag of the bot. The visual tag indicates the bot's deployment stage.

Returns:

str | None: The visual tag (e.g., "dev", "staging", "prod") or None if not set.

get_config_value​

def get_config_value(key: str) -> str | None

Get a configuration value from the bot's configmap.

Arguments:

• key - The configuration key to retrieve

Returns:

The configuration value as a string, or None if the key doesn't exist

get_config_map​

def get_config_map() -> dict[str, str]

Get the entire configuration map for the bot.

Returns:

A dictionary containing all configuration key-value pairs

shared.tracker.apis._state_manipulation_api

StateManipulationApiConfig Objects​

class StateManipulationApiConfig(AuthorizedApiConfig)

Configuration for state manipulation APIs with conversation state.

StateManipulationApi Objects​

class StateManipulationApi(AuthorizedApi)

API for managing conversation state synchronization.

Provides state sync capabilities and guarded access to ensure state consistency between agent and LLM service.

guarded_state​

@staticmethod
def guarded_state(func: Callable) -> Callable

Decorator that ensures state is synchronized from the agent service before method execution.

This decorator checks if sync_state_id is None and if so, queries the state from the agent service and updates the tracker's state before executing the decorated method.

Arguments:

• func - The method to be decorated

Returns:

The wrapped method that ensures state synchronization

sync_state​

async def sync_state()

Sync the state from the agent service. This method is used to sync the state from the agent service. The state is synced as soon as this method is called.

shared.tracker.apis._periodic_script_api

PeriodicScriptTrackerApiConfig Objects​

class PeriodicScriptTrackerApiConfig(LLMCallApiConfig)

Configuration for periodic script tracker APIs. Currently copy of LLMCallApiConfig.

We use this custom class though to easier update behaviour and config in the future.

PeriodicScriptTrackerApi Objects​

class PeriodicScriptTrackerApi(LLMCallApi, ConfigMapApi)

Tracker API for periodic scripts with LLM access and configmap helpers.

shared.tracker.phone

shared.tracker.base

shared.tracker._phone

shared.tracker._state

shared.tracker.state

shared.tracker._base

shared.tracker._api

shared.tracker.api

shared.conversation

shared.custom_events

shared.custom_events._platform_events

shared.custom_events.types

shared.custom_events._utterance_events

LLMStreamedUtteranceStartingEventPayload Objects​

class LLMStreamedUtteranceStartingEventPayload(
        LLMStreamedUtteranceStartingBaseEventPayload)

llm_config​

typing the llm_config as LLMAPIConfig

LLMStreamedUtteranceRunningEventPayload Objects​

class LLMStreamedUtteranceRunningEventPayload(
        LLMStreamedUtteranceRunningBaseEventPayload)

LLMStreamedUtteranceRunningBaseEventPayload indicates that a streamed utterance has already started and is currently running.

Attributes:

• text str | None - The text of the current chunk.
• type SupportedProviders - The type of the provider that is being used.
• status Literal["running"] - The status of the streamed utterance event, set to "running".
• chunk ModelResponseStream - The current chunk of the chat response.

LMMUtteranceEventPayload Objects​

class LMMUtteranceEventPayload(LMMUtteranceBaseEventPayload)

llm_config​

typing the llm_config as LLMAPIConfig

shared.custom_events._story_state_events

LLMInferenceEventPayload Objects​

class LLMInferenceEventPayload(LLMInferenceBaseEventPayload)

llm_config​

typing the llm_config as LLMAPIConfig

LLMInferenceEvent Objects​

class LLMInferenceEvent(LLMInferenceBaseEvent, CustomPlatformEvent)

payload​

using the typed payload

shared.custom_events.platform_events

shared.custom_events._orchestrator_events

shared.custom_events._types

ParsedEvent Objects​

class ParsedEvent(BaseModel)

event​

type: ignore

shared.custom_events._utils

shared.custom_events.orchestrator_events

shared.custom_events.utterance_events

shared.custom_events.utils

shared.custom_events.story_state_events

shared.execution_utils

shared.execution_utils.auth

shared.execution_utils._requests

InputGenerationAgentRequest Objects​

class InputGenerationAgentRequest(RemoteExecutionCall)

Request to evaluate a custom condition

ConditionCall Objects​

class ConditionCall(RemoteExecutionCall)

Request to evaluate a custom condition

LlmJudgeCall Objects​

class LlmJudgeCall(RemoteExecutionCall)

Request to evaluate an LLM judge condition

Variable overrides:

• Reserved variables (evaluation_criterium, temperature, history, provider_id) are applied in remote_llm_judge_executor.py before creating this call
• Custom template variables are stored in variables dict and passed to the prompt template rendering

Note: The custom prompt template is available in llm_api_config.prompt

evaluation_criterium​

What to evaluate

variables​

Custom template variables for prompt rendering

PeriodicScriptCall Objects​

class PeriodicScriptCall(BaseModel)

Request to execute a periodic Python script. The script is executed in the bot's action container context.

shared.execution_utils.remote_execution

shared.execution_utils._utils

shared.execution_utils._remote_execution

shared.execution_utils.requests

shared.execution_utils.responses

shared.execution_utils._responses

ConditionSuccessStatus Objects​

class ConditionSuccessStatus(BaseModel)

message​

Custom message from script or reasoning from LLM judge

ConditionFailureStatus Objects​

class ConditionFailureStatus(BaseModel)

message​

Custom message from script or reasoning from LLM judge

PeriodicScriptResponse Objects​

class PeriodicScriptResponse(BaseModel)

Response from executing a periodic Python script.

Contains the execution result, combined output (logs + stdout + stderr) and execution time.

output​

combined output: logs, stdout, stderr + stacktrace

shared.execution_utils.utils

shared.execution_utils._local_execution

shared.execution_utils.local_execution

shared.execution_utils._auth

create_auth_token​

def create_auth_token(bot_id: str,
                      tenantId: str,
                      roles: list[str] | None = None,
                      session_id: str | None = None) -> SecretStr