Custom Ittybit component for Langflow

View Markdown

Langflow lets you wire AI workflows together visually, but it has no built-in media processing. By creating a custom Ittybit component, you get a drag-and-drop node that can transcode video, extract audio, generate thumbnails, and create HLS streams — all from within the Langflow canvas.

Install dependencies

You need requests available in your Langflow environment. If you’re running Langflow locally:

pip install requests

Create the component

In Langflow, open the code editor for a new custom component. Paste the full component class below. It POSTs a task to the Ittybit API, polls until completion, and returns the output URL.

import time
from typing import Optional

import requests
from langflow.custom import Component
from langflow.io import DropdownInput, MessageTextInput, Output, SecretStrInput
from langflow.schema import Data


class IttybitTask(Component):
    display_name = "Ittybit Media Task"
    description = "Process media files using the Ittybit Tasks API"
    icon = "video"

    inputs = [
        SecretStrInput(
            name="api_key",
            display_name="API Key",
            info="Your Ittybit API key",
            required=True,
        ),
        MessageTextInput(
            name="input_url",
            display_name="Input URL",
            info="URL of the source media file",
            required=True,
        ),
        DropdownInput(
            name="kind",
            display_name="Task Kind",
            options=["video", "audio", "image", "adaptive_video"],
            value="video",
            info="Type of media processing task",
        ),
        DropdownInput(
            name="format",
            display_name="Output Format",
            options=["mp4", "webm", "mp3", "aac", "opus", "webp", "png", "jpg", "avif"],
            value="mp4",
            info="Desired output format",
            advanced=True,
        ),
        MessageTextInput(
            name="width",
            display_name="Width",
            info="Output width in pixels (optional)",
            advanced=True,
        ),
        MessageTextInput(
            name="height",
            display_name="Height",
            info="Output height in pixels (optional)",
            advanced=True,
        ),
        DropdownInput(
            name="quality",
            display_name="Quality",
            options=["low", "medium", "high"],
            value="medium",
            info="Quality preset",
            advanced=True,
        ),
    ]

    outputs = [
        Output(display_name="Result", name="result", method="run"),
    ]

    def run(self) -> Data:
        # Build the task payload
        options: dict = {
            "format": self.format,
            "quality": self.quality,
        }
        if self.width:
            options["width"] = int(self.width)
        if self.height:
            options["height"] = int(self.height)

        payload = {
            "input": self.input_url,
            "kind": self.kind,
            "options": options,
        }

        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
        }

        # Create the task
        res = requests.post(
            "https://api.ittybit.com/jobs",
            headers=headers,
            json=payload,
        )
        res.raise_for_status()
        task = res.json()
        task_id = task["id"]

        # Poll until the task completes
        result = self._poll(task_id, headers)
        return Data(data=result)

    def _poll(
        self,
        task_id: str,
        headers: dict,
        timeout: int = 300,
        interval: int = 2,
    ) -> dict:
        deadline = time.time() + timeout
        while time.time() < deadline:
            res = requests.get(
                f"https://api.ittybit.com/jobs/{task_id}",
                headers=headers,
            )
            res.raise_for_status()
            data = res.json()

            if data["status"] == "completed":
                return data
            if data["status"] == "error":
                raise RuntimeError(
                    f"Task {task_id} failed: {data.get('error', 'unknown error')}"
                )

            time.sleep(interval)

        raise TimeoutError(f"Task {task_id} did not complete within {timeout}s")

How it works

The component exposes seven inputs that map directly to the Ittybit task body:

API Key — stored as a secret, never exposed in the canvas
Input URL — the source media file
Task Kind — video, audio, image, or adaptive_video
Output Format, Width, Height, Quality — advanced options that control the output

When the flow runs, the run method POSTs to POST /jobs, then polls GET /jobs/:id every 2 seconds until the job reaches completed or error. The full task response (including output URLs) is returned as a Data object that downstream nodes can consume.

Use it in a flow

Once the component is saved, it appears in Langflow’s sidebar. A typical flow looks like:

Chat Input — user provides a media URL and instructions
LLM node — parses the request and extracts the URL, kind, and options
Ittybit Media Task — processes the file
Chat Output — returns the result URL to the user

Connect the LLM output to the Ittybit component’s Input URL field, and wire the result into Chat Output.

Fire and forget variant

If you don’t need to wait for the result inside Langflow (maybe you’re using webhooks instead), strip out the polling:

def run(self) -> Data:
    options: dict = {
        "format": self.format,
        "quality": self.quality,
    }
    if self.width:
        options["width"] = int(self.width)
    if self.height:
        options["height"] = int(self.height)

    res = requests.post(
        "https://api.ittybit.com/jobs",
        headers={
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
        },
        json={
            "input": self.input_url,
            "kind": self.kind,
            "options": options,
        },
    )
    res.raise_for_status()
    return Data(data=res.json())