Code Component (logic.code)

The logic.code component executes custom Python code safely within the workflow execution environment. It provides a sandboxed space equipped with helpers for input/output data handling and flow control.

Runtime Environment

The execution occurs inside a hardened Python 3 environment.

Available Modules

You can leverage several standard libraries and helpful built-ins: Standard Libraries:

json, uuid, time, datetime, zoneinfo
functools, traceback, copy, decimal
textwrap, math, re, abc
typing, Union, Literal

Exceptions:

Exception
ValueError
Exit: A special exception intentionally designed to exit the active code block immediately.

Session exceptions: raised by the Session Management functions below, and catchable in your code so a workflow can react — most importantly, retry an optimistic-concurrency clash on a later run.

InvalidSessionObjectAlias: An alias violates the namespace grammar.
InvalidThreadEnvelope / InvalidGraphEnvelope: The target object is missing, or is not a thread / graph.
GraphMutationConflict: An if_rev optimistic-lock check failed — the element changed since it was read.
GraphTraversalLimitExceeded: A query_session_graph traversal exceeded a hard bound.
InvalidGraphQuery: A malformed traversal query.

Input & Output

Reading Inputs

You can access data arriving at the component’s input ports. Inputs are received as explicitly typed Value Objects.

data_inputs: A list of all received input port data.
get_data_input(route): A helper function to grab the Value Object of a specific input port by its exact route name.

# Returns a dict structurally conforming to a Value Object.
input_val = get_data_input("user_query")

if input_val and input_val.get("type") == "message":
    pass

Writing Outputs

You can emit data out of the component by pushing Value Objects to output routes.

set_data_output(route, value, persist=False): Assigns a value to a targeted output port.
- route: Output port name (e.g., “result”).
- value: The Value Object payload.
- persist: If True, stores the output specifically to processor data tracking logic for external visibility.

output_val = create_message(role="assistant", text="Processed Data")
set_data_output("result", output_val)

Flow Control

You determine which logic route the workflow should navigate following code execution.

set_result(route, terminal=False): Declares the destination route.
- route: Connection edge to traverse ("success", "failure", etc).
- terminal: Set to True if this intentionally ends the entire workflow execution.
exit(): Aborts the execution of the remainder of the active code script.

if user_exists:
    set_result("success")
else:
    set_result("failure")

Helper Functions

Data Helpers

create_message(role, text=None, attribute_id=None): Generates a standardized “message” Value Object dynamically.
create_value(data_type, data): Quickly wrap arbitrary matching data into a recognized Value structure.

Session Management

Every function below is bound to the execution’s own session — none of them takes a session_id, so your code can only ever reach its own session’s objects. Names and behavior mirror the platform’s REST session commands one-to-one, and the same validation applies here as on the REST API (alias grammar, thread/graph envelope checks, size limits, graph mutation/traversal caps, and if_rev optimistic concurrency). Writes are attributed to the user the execution runs as. The request/response shapes for each function are documented in full on the linked operations reference pages below — this section covers the signature, behavior, and return shape only.

Objects

A session object is a typed document addressed by an alias — a path such as missions/a1/graph.

download_session_object(alias): Reads the value of a session object. Returns the typed payload, or None if it doesn’t exist. See Download Session Object.
upload_session_object(alias, value): Writes/creates a session object, validating the alias. See Upload Session Object.
delete_session_object(alias): Deletes a session object. If the object is a thread or graph envelope, this also cascades — deleting all of its child thread items / graph elements. See Session Objects for the envelope model (there’s no dedicated delete reference page).
list_session_objects(prefix="", delimiter=None, cursor=None, limit=None): Lists session objects under an optional alias prefix, with optional delimiter-based folder rollup and cursor pagination. Returns {"objects": [...], "cursor": ..., "common_prefixes": [...]} (the last key only when delimiter is given). See List Session Objects.

config = download_session_object("my_config")                     # read
upload_session_object("counter", {"count": 1})                     # write
listing = list_session_objects(prefix="missions/", delimiter="/")  # one-level "folder" listing

Threads

A thread object holds an append-only list of items. Create the envelope first with upload_session_object(alias, {"type": "thread", ...}) before posting to it.

post_session_thread_item(alias, content, parent_id=None, metadata=None): Appends an item. content is a list of typed parts (e.g. [{"type": "text", "text": "..."}]); metadata is a free-form tag dict. Returns {"item": <item>}. See Post Session Thread Item.
list_session_thread_items(alias, ascending=False, created_since=None, created_before=None, cursor=None, limit=None): Reads items, newest-first by default; created_since/created_before bound the range (use created_since as a delta cursor). Returns {"items": [...], "cursor": ...}. See List Session Thread Items.
get_session_thread_item(alias, item_id): Fetches a single item. Returns {"item": <item>}. See Get Session Thread Item.

posted = post_session_thread_item("missions/a1/cells/imp-7f/meta",
                                   [{"type": "text", "text": "Plan ready."}],
                                   metadata={"type": "summary"})
recent = list_session_thread_items("missions/a1/cells/imp-7f/meta", created_since=cursor)

Graphs

A graph object holds vertices and edges. Create the envelope first with upload_session_object(alias, {"type": "graph", ...}) — vertices and edges are then changed only via mutations.

apply_session_graph_mutations(alias, operations): Applies a batch of mutations (add_vertex, add_edge, set_vertex_props, set_edge_props, remove_*_props, delete_vertex, delete_edge), each optionally carrying if_rev for optimistic concurrency. Returns {"elements": [...], "changes": [...]}. See Apply Session Graph Mutations.
query_session_graph(alias, query): Runs a bounded traversal against a query AST. Returns {"result": ...}. See Query Session Graph.
list_session_graph_elements(alias, type=None, ascending=False, updated_since=None, ids_only=False, cursor=None, limit=None): Pages elements, optionally filtered to type "vertex"/"edge". Returns {"elements": [...], "cursor": ...}, or {"element_ids": [...], "cursor": ...} when ids_only=True. See List Session Graph Elements.
get_session_graph_element(alias, element_id): Fetches one element. Returns {"element": ...}. See Get Session Graph Element.
get_session_graph_elements(alias, element_ids): Batch-fetches elements. Returns {"elements": [...]}. See Get Session Graph Elements.
list_session_graph_neighbors(alias, from_ids, direction, label=None, neighbor_label=None, cursor=None, limit=None): Expands adjacency from from_ids (direction is "out"/"in"/"both", optionally filtered by edge label / neighbor_label). Returns {"edges": [...], "vertices": [...], "cursor": ...}. See List Session Graph Neighbors.

mutated = apply_session_graph_mutations("missions/a1/graph", [
    {"op": "add_vertex", "type": "cell", "props": {"status": "planned"}},
])
open_cells = query_session_graph("missions/a1/graph", {
    "start": ["V"], "steps": [["hasLabel", "cell"], ["has", "status", "neq", "completed"]],
})

Advanced: optimistic concurrency

Mutations can carry if_rev to guard against a concurrent write. If another writer already advanced the element since it was read, the call raises GraphMutationConflict instead of silently overwriting it — catch the exception and reconcile on a later run:

try:
    apply_session_graph_mutations("missions/a1/graph", [
        {"op": "set_vertex_props", "element_id": "imp-7f", "props": {"status": "working"}, "if_rev": 4},
    ])
except GraphMutationConflict:
    pass  # another writer advanced it; reconcile on the next run

Logging & Events

Produce informational, warning, or failure events logged to the execution stream:

post_output_event(message)
post_warning_event(message)
post_error_event(message)

System Metadata

current_time_utc_iso(): Gets UTC time dynamically.
result: The immediate runtime representation of the result dictionary.

Full Example

request_val = get_data_input("request")

if request_val and request_val.get("type") == "message":
    
    response_text = "Message received and acknowledged."
    
    msg = create_message("assistant", response_text)
    set_data_output("response", msg)
    
    post_output_event("Ack sent")
    set_result("success")
else:
    post_error_event("Request generic or missing type string")
    set_result("failure")

Overview

Workflow

Scheduler

ONNX

Code Component (logic.code)

Runtime Environment

Available Modules

Input & Output

Reading Inputs

Writing Outputs

Flow Control

Helper Functions

Data Helpers

Session Management

Objects

Threads

Graphs

Advanced: optimistic concurrency

Logging & Events

System Metadata

Full Example

​Runtime Environment

​Available Modules

​Input & Output

​Reading Inputs

​Writing Outputs

​Flow Control

​Helper Functions

​Data Helpers

​Session Management

​Objects

​Threads

​Graphs

​Advanced: optimistic concurrency

​Logging & Events

​System Metadata

​Full Example

Runtime Environment

Available Modules

Input & Output

Reading Inputs

Writing Outputs

Flow Control

Helper Functions

Data Helpers

Session Management

Objects

Threads

Graphs

Advanced: optimistic concurrency

Logging & Events

System Metadata

Full Example