Building Langflow Components — Safe, Tested, Forever-Named

Summary. This is a niche partner integration, and the acme bundle already exists. So the component goes into that bundle and needs no new bundle or RFC. The class name AcmeModerationComponent and the input and output names below are permanent: please confirm them before this merges. The component calls an AI service, so it has an explicit timeout, maps HTTP failures to clear messages on the node, and never logs the text or the key.

Placement and names (decide once)

File: src/bundles/acme/src/lfx_acme/components/acme/acme_moderation.py. Before creating it, check that no existing Acme component already wraps /v1/check.

• Class: AcmeModerationComponent, following <Provider><Capability>Component.
• Inputs: api_key, input_value, timeout_seconds. Output: moderation_result.

Icon: "shield-check" (Lucide). If AcmeChatModelComponent uses a custom Acme SVG, use that instead so the bundle stays consistent.

Component

import time
from typing import NoReturn

from langflow.custom import Component
from langflow.io import IntInput, MessageTextInput, Output, SecretStrInput
from langflow.schema import Data
from loguru import logger

MODERATION_URL = "https://moderation.acme.example/v1/check"
STATUS_MESSAGES = {
    401: "Acme rejected the API key. Check your secret.",
    429: "Acme rate limit reached. Try again later.",
}

def _status_message(status_code: int) -> str:
    return STATUS_MESSAGES.get(status_code, f"Acme Moderation failed: HTTP {status_code}")

class AcmeModerationComponent(Component):
    display_name = "Acme Moderation"
    description = "Check text with Acme's moderation API and return whether it was flagged."
    icon = "shield-check"
    documentation = "https://moderation.acme.example/docs"

    inputs = [
        SecretStrInput(
            name="api_key",
            display_name="API Key",
            info="Acme Moderation API key. Stored encrypted.",
            required=True,
        ),
        MessageTextInput(name="input_value", display_name="Text", required=True),
        IntInput(name="timeout_seconds", display_name="Timeout (s)", value=10, advanced=True),
    ]

    outputs = [
        Output(display_name="Moderation Result", name="moderation_result", method="check_text"),
    ]

    def check_text(self) -> Data:
        import httpx

        start = time.monotonic()
        try:
            response = httpx.post(
                MODERATION_URL,
                headers={"Authorization": f"Bearer {self.api_key}"},
                json={"text": self.input_value},
                timeout=self.timeout_seconds,
            )
            response.raise_for_status()
        except httpx.TimeoutException as exc:
            self._fail(f"Acme Moderation timed out after {self.timeout_seconds}s.", "timeout", start, exc)
        except httpx.HTTPStatusError as exc:
            self._fail(_status_message(exc.response.status_code), "http_error", start, exc)
        except httpx.RequestError as exc:
            self._fail("Could not reach Acme Moderation.", "unreachable", start, exc)

        body = response.json()
        logger.info(
            "acme.moderation",
            outcome="success",
            text_chars=len(self.input_value),
            duration_ms=int((time.monotonic() - start) * 1000),
        )
        return Data(data={"flagged": bool(body["flagged"]), "categories": list(body.get("categories", []))})

    def _fail(self, message: str, outcome: str, start: float, exc: Exception) -> NoReturn:
        logger.warning(
            "acme.moderation",
            outcome=outcome,
            duration_ms=int((time.monotonic() - start) * 1000),
        )
        self.status = message
        raise ValueError(message) from exc

httpx is imported inside the method, so loading the bundle never fails because of it. The logs record the outcome, the text length and the duration. They never record the text, the response or the key. There are no retries: none are implemented by hand, and with no Acme SDK there is nothing to delegate them to. A flow that needs retries can add them around the node.

Registration and changelog

• Add AcmeModerationComponent to src/bundles/acme/src/lfx_acme/components/acme/__init__.py, keeping the alphabetical order.
• Add httpx to the bundle's pyproject.toml with a compatible range, unless langflow-base already provides it.
• In src/bundles/acme/BUNDLE_API.md, under ## [Unreleased] → ### Added: `AcmeModerationComponent`: check text with Acme's moderation API (`/v1/check`).

Tests (src/bundles/acme/tests/unit/components/test_acme_moderation.py)

import os

import httpx
import pytest
from langflow.schema import Data
from lfx_acme.components.acme.acme_moderation import AcmeModerationComponent

from tests.base import ComponentTestBaseWithoutClient

class TestAcmeModerationComponent(ComponentTestBaseWithoutClient):
    @pytest.fixture
    def component_class(self):
        return AcmeModerationComponent

    @pytest.fixture
    def default_kwargs(self):
        return {"api_key": "test-key", "input_value": "Hello there", "timeout_seconds": 10}

    @pytest.fixture
    def file_names_mapping(self):
        return {}

    @pytest.mark.api_key_required
    def test_should_return_moderation_data_when_called_against_real_api(self, component_class, default_kwargs):
        default_kwargs["api_key"] = os.environ["ACME_MODERATION_API_KEY"]
        result = component_class(**default_kwargs).check_text()
        assert isinstance(result, Data)
        assert isinstance(result.data["flagged"], bool)

    def test_should_raise_value_error_when_request_times_out(self, component_class, default_kwargs, mocker):
        mocker.patch("httpx.post", side_effect=httpx.ReadTimeout("slow"))
        component = component_class(**default_kwargs)
        with pytest.raises(ValueError, match="timed out after 10s"):
            component.check_text()
        assert "timed out" in component.status

    def test_should_report_rate_limit_when_api_returns_429(self, component_class, default_kwargs, mocker):
        request = httpx.Request("POST", "https://moderation.acme.example/v1/check")
        mocker.patch("httpx.post", return_value=httpx.Response(429, request=request))
        component = component_class(**default_kwargs)
        with pytest.raises(ValueError, match="rate limit"):
            component.check_text()

The real API test is the main check. Mocks cover only the timeout and the 429, which a sandbox can't reliably reproduce, as Langflow's testing policy recommends. The base class's automatic tests (instantiation, inputs, outputs) run from the three fixtures.

Commands to run (not executed here)

LFX_DEV=acme make backend      # the component should appear in the sidebar; drag it onto a flow
uv sync --group dev --package lfx-acme
uv run pytest src/bundles/acme/tests/ -v
make format_backend && uv run git commit

The expected outcome: the two mocked tests pass, and the api_key_required test is skipped unless ACME_MODERATION_API_KEY is set. None of this has been run yet.

Next steps

1. Confirm the class, input and output names. After the merge they can't change.
2. Run the commands above. Check that the node renders and that a 401 shows "Acme rejected the API key" on the node.
3. Learning capture: nothing new beyond SKILL.md and AGENTS.md came up, so no learnings/ entry was written.