Skip to content

Tutorial: Creating Custom Flows in Kelvin Cortex#

Last updated: February 2026

Important current limitation (February 2026)
Custom Flows in Kelvin Cortex currently support only JSON-serialisable return values.

This tutorial explains how to write a Cortex-compatible function and release it as a new custom Flow.

1. What You Can (and Should) Return#

Valid return types include:

  • dict / Dict[str, Any]
  • list / List[Any]
  • float, int, str, bool, None
  • Nested combinations of the above

Common patterns:

  • Per-frame or per-timestamp metrics
  • Aggregated statistics (mean, max, min, counts…)
  • Detected events / timestamps
  • Keypoint-based features (positions, distances, angles…)
  • Classification results / scores

2. Basic Flow Function Structure#

from typing import Dict, Any, List
import numpy as np
# Assume these types are available in your Cortex environment
from cortex.types import VideoObject, PoseObject

def compute_wrist_symmetry(
    pose: PoseObject
) -> Dict[str, Any]:
    """
    Computes left-right wrist symmetry metrics over time.
    Returns only JSON-compatible data.
    """

    # Extract relevant data
    left  = pose.pose_array["left_wrist"][:, :2]   # (T, 2) → x,y
    right = pose.pose_array["right_wrist"][:, :2]
    conf_l = pose.pose_array["left_wrist"][:, -1]
    conf_r = pose.pose_array["right_wrist"][:, -1]

    # Only consider frames where both wrists are reasonably confident
    valid = pose.valid_times & (conf_l > 0.4) & (conf_r > 0.4)

    distances = np.linalg.norm(left - right, axis=1)[valid]
    timestamps_ms = pose.timestamps[valid]

    # Basic statistics
    stats = {
        "mean_distance_px": float(np.mean(distances)),
        "std_distance_px": float(np.std(distances)),
        "min_distance_px": float(np.min(distances)),
        "max_distance_px": float(np.max(distances)),
        "num_valid_frames": int(np.sum(valid)),
        "total_frames": len(pose.timestamps),
    }
    return stats

3. Publishing with FlowBuilder#

3.1 Obtaining an API token#

To generate an API token that will allow you to publish custom Flows visit the Cortex Flows page.

Copy the generated string and initialise a FlowBuilder instance with it. The token will be valid for 24h.

3.2 Using FlowBuilder to publish a Flow#

from cortex.flow import FlowBuilder

builder = FlowBuilder(api_token="your-api-token-here")

# Validate first (recommended)
serialised, spec = builder.build(compute_wrist_symmetry)

# Publish
response = builder.publish(
    func=compute_wrist_symmetry,
    flow_display_name="Wrist Symmetry Analyser",
    flow_description="Computes left-right wrist distance metrics.",
    output_display_name="Wrist Symmetry JSON Report",
    output_description="Dictionary containing statistics and per-frame distances."
)

The published Flow will only be visible to the user whose token has been used to publish it!

4. Final Notes#

For now, custom functions can be used to compute input-based metrics or timeseries, but their output is limited to standard Python types. Future platform updates may lift some of the I/O restrictions — check Cortex documentation or changelogs. Until then: focus on quantitative analysis, event detection, feature extraction, and structured reporting.