Code Component (logic.code)

Execute custom Python code within the workflow environment.

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.
• LimitExceeded: A write would create a new session object past the execution’s workflow.session_objects_per_session. It carries the limit object. See Limits.

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.

Limits

Four limits shape a logic.code run. All four come from the session’s blob, and each is fixed when the execution is created.

• workflow.code_run_seconds — 870 seconds by default, and at most. A run still going at that point is stopped; the node takes its failure route, and the error message names the limit.
• workflow.code_imports — the third-party modules the code may import beyond the standard libraries above, empty by default. A module on the list is imported when the program names it, and is bound under its own name. The entry "*" makes every listed module import before the code runs, whether the program names it or not; it adds no module the list does not name.
• workflow.processor_runs_per_execution — each logic.code run is one node run, so a loop through a code node counts every pass.
• workflow.session_objects_per_session — a write that would create a new session object past it raises LimitExceeded, carrying the limit object. The check uses the value the execution started with, where a REST write uses the value in force when it arrives.

A write past a fixed size or count limit fails here as its REST command does.

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")