Python Node¶

The Python node allows the bot builder to execute custom Python code to perform logic, data processing, or other tasks.

The code must define a main function which takes the node input as a string and returns a string to pass to the next node. The main function must also accept arbitrary keyword arguments to support future features. Here is an example of what the code might look like:

def main(input, **kwargs) -> str:
    # Put your code here
    return input

The input parameter is a string that contains the input to the node. The return value of the function is a string that will be passed to the next node in the pipeline.

Additional Keyword Arguments¶

The following additional arguments are provided:

node_inputs: list[str] - A list of all the inputs to the node at the time of execution. This will be the same as [input] except when the node is part of a workflow with parallel branches.

Warning

All the code must be encapsulated in a main function. You can write other functions but they must be within the scope of the main function. For example:

def main(input, **kwargs):
    def important(arg):
        return arg + "!"

    return important(input)

Utility Functions¶

The Python node provides a set of utility functions that can be used to interact with the user's data and the pipeline state.

`get_participant_data() -> dict` ¶

Returns the current participant's data as a dictionary.

`set_participant_data(data: dict) -> None` ¶

Updates the current participant's data with the provided dictionary. This will overwrite any existing data.

`set_participant_data_key(key_name: str, data: any) -> None` ¶

Updates the current participant's data with the provided value at the specified key.

`append_to_participant_data_key(key_name: str, data: any) -> None` ¶

Appends the provided value to the participant's data at the specified key. If the value at the key is not a list, it will be converted to a list containing the provided value.

`increment_participant_data_key(key_name: str, data: any) -> None` ¶

Increments the value at the participant's data key with the specified value

`get_participant_schedules() -> list` ¶

Returns all active scheduled messages for the participant in the current experiment session.

`get_temp_state_key(key_name: str) -> str | None` ¶

Returns the value of the temporary state key with the given name. If the key does not exist, it returns None.

`set_temp_state_key(key_name: str, data: Any) -> None` ¶

Sets the value of the temporary state key with the given name to the provided data. This will override any existing data for the key.

`get_session_state_key(key_name: str) -> str | None` ¶

Returns the value of the session state key with the given name. If the key does not exist, it returns None.

`set_session_state_key(key_name: str, data: Any) -> None` ¶

Sets the value of the session state key with the given name to the provided data. This will override any existing data for the key.

`get_selected_route(router_node_name: str) -> str | None` ¶

Returns the route selected by a specific router node with the given name. If the node does not exist or has no route defined, it returns None.

`get_node_path(node_name: str) -> list | None` ¶

Returns a list containing the sequence of nodes leading to the target node. If the node is not found in the pipeline path, returns a list containing only the specified node name.

`get_all_routes() -> dict` ¶

Returns a dictionary containing all routing decisions made in the pipeline up to the current node. The keys are the node names and the values are the route keywords chosen by each router node.

Note that in parallel workflows only the most recent route for a particular node will be returned.

`add_message_tag(tag_name: str)` ¶

Adds a tag to the output message. To add multiple tags, call this function multiple times.

`add_session_tag(tag_name: str)` ¶

Adds a tag to the chat session. To add multiple tags, call this function multiple times.

`get_node_output(node_name: str) -> Any` ¶

Returns the output of the specified node if it has been executed. If the node has not been executed, it returns None.

`require_node_outputs(*node_names)` ¶

This function is used to ensure that the specified nodes have been executed and their outputs are available in the pipeline's state. If any of the specified nodes have not been executed, the node will not execute and the pipeline will wait for the required nodes to complete.

This should be called at the start of the main function.

def main(input, **kwargs):
    require_node_outputs("nodeA", "nodeB")
    return get_node_output("nodeA") + get_node_output("nodeB")

`wait_for_next_input()` ¶

Advanced utility that will abort the current execution when not all inputs have been received. This is only useful in cases where the workflow has parallel branches which might result in the node being executed more than once.

This is similar to require_node_outputs but useful where some node outputs may be optional.

def main(input, **kwargs):
    a = get_node_output("a")
    b = get_node_output("b")
    if not a and not b:
        wait_for_next_input()
    # do something with a or b

`abort_with_message(message, tag_name: str = None) -> None` ¶

Calling this will terminate the pipeline execution. No further nodes will get executed in any branch of the pipeline graph.

The message provided will be used to notify the user about the reason for the termination. If a tag name is provided, it will be used to tag the output message.

# Abort pipeline with custom message
if safety_violation_detected:
    abort_with_message("Content policy violation detected")

`attach_file_from_response(response_bytes: bytes, filename: str) -> None` ¶

Attaches a file downloaded from an HTTP response to the chat session.

This function is used in combination with the HTTP client to download files from external APIs and attach them to the assistant's response message. The file will be available for the user to download.

Parameters:

Name	Type	Description	Default
`response_bytes`	`bytes`	The raw bytes of the file, typically from the `response_bytes` field of an HTTP response	required
`filename`	`str`	The name to give the attached file, including the file extension	required

Example

def main(input, **kwargs):
    # Download a file from an API
    response = http.get("https://api.example.com/report.pdf", auth="my-api")

    if response["is_success"]:
        # Attach the file to the chat
        attach_file_from_response(
            response_bytes=response["response_bytes"],
            filename="report.pdf"
        )
        return "I've attached the report for you."

    return "Failed to download the report."

HTTP Client¶

The Python node provides an http global that enables secure HTTP requests to external APIs. See the HTTP Client page for full documentation.

Temporary State¶

The Python node can also access and modify the temporary state of the pipeline. The temporary state is a dictionary that is unique to each run of the pipeline (each new message from the user) and is not stored between sessions.

The temporary state can be accessed and modified using the get_temp_state_key and set_temp_state_key utility functions.

Temporary state contains the following keys by default. These keys can not be modified or deleted:

Key	Description
`user_input`	The message sent by the user
`outputs`	The outputs generated by the previous node
`attachments`	A list of attachments passed in by the user. See Attachments

In addition to these keys, the temporary state can also contain custom key-value pairs that can be set and accessed by the Python node and by the Static Router node.

Here is an example of a temporary state dictionary:

{
    "user_input": "Please help me with my problem",
    "outputs": {
        "Assistant": "I'm here to help! What can I do for you?"
    },
    "attachments": [
        Attachment(...),
    ],
    "my_custom_key": "my_custom_value",
}

Session State¶

The Python node can also access and modify the state of the participant's session. This state is a dictionary that is scoped to each session that the user might have with the bot.

The session state can be accessed and modified using the get_session_state_key and set_session_state_key utility functions.

Attachments¶

Part of the temporary state is a list of attachments. Attachments are files that the user has uploaded to the bot. Each attachment has the following fields:

Field	Description
`name`	The name of the file
`size`	The size of the file in bytes
`content_type`	The MIME type of the file
`upload_to_assistant`	Whether the file should be uploaded to the assistant as an attachment
`read_bytes()`	Reads the attachment content as bytes.
`read_text()`	Reads the attachment content as text.

Here is an example of an attachment object:

attachment = Attachment(
    name="proposal.pdf",
    size=1234,
    content_type="application/pdf",
    upload_to_assistant=False,
)
content = attachment.read_text()

Supported File Types¶

The Python node currently only supports reading the contents of the following file types:

Text-based formats (TXT, CSV, HTML, JSON, XML, etc.)
PDF
DOCX
XLSX
XLS
Outlook
PPTX

Other file types can still be uploaded to assistants but the Python Node is not able to read the file contents using the read_text() method on the attachment.

Python Node¶

Additional Keyword Arguments¶

Utility Functions¶

get_participant_data() -> dict ¶

set_participant_data(data: dict) -> None ¶

set_participant_data_key(key_name: str, data: any) -> None ¶

append_to_participant_data_key(key_name: str, data: any) -> None ¶

increment_participant_data_key(key_name: str, data: any) -> None ¶

get_participant_schedules() -> list ¶

get_temp_state_key(key_name: str) -> str | None ¶

set_temp_state_key(key_name: str, data: Any) -> None ¶

get_session_state_key(key_name: str) -> str | None ¶

set_session_state_key(key_name: str, data: Any) -> None ¶

get_selected_route(router_node_name: str) -> str | None ¶

get_node_path(node_name: str) -> list | None ¶

get_all_routes() -> dict ¶

add_message_tag(tag_name: str) ¶

add_session_tag(tag_name: str) ¶

get_node_output(node_name: str) -> Any ¶

require_node_outputs(*node_names) ¶

wait_for_next_input() ¶

abort_with_message(message, tag_name: str = None) -> None ¶

attach_file_from_response(response_bytes: bytes, filename: str) -> None ¶