# Custom Ittybit component for Langflow

Build a drag-and-drop media processing node for Langflow using the Ittybit API

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:

```bash
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.

```python

from typing import Optional

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:

1. **Chat Input** -- user provides a media URL and instructions
2. **LLM node** -- parses the request and extracts the URL, kind, and options
3. **Ittybit Media Task** -- processes the file
4. **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](/docs/webhooks) instead), strip out the polling:

```python
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())
```

This returns immediately with the task ID and `queued` status. Your webhook endpoint picks up the result when it's done.

## See also

- [Ittybit Task API reference](/reference/tasks)
- [Create HLS streams](/guides/create-hls-streams)
- [Langflow custom components docs](https://docs.langflow.org/components-custom-components)