Test the Langflow run endpoint as a real contract

The flow answers correctly from the Langflow canvas, but the service client gets a validation response from /api/v1/run/.... After the payload is “fixed,” the endpoint returns 200 and the application still reads the wrong output branch. The canvas run and the external API contract were never the same test.

A reliable check has to cover the request shape, the deployed flow, and the response value your caller consumes. Status codes are necessary evidence, but they do not prove that the intended input reached the intended component or that the client extracted the intended output.

Start with the contract exposed by the deployed flow

Current Langflow documentation describes POST /api/v1/run/{flow_id_or_name} as the simplified flow trigger. Its request accepts top-level input_value as an optional field with a default of null, alongside optional input_type, output_type, output_component, tweaks, and session_id. A particular flow, such as the documented Basic Prompting example, can still require usable chat input. The documentation lists chat as the default for input_type and output_type, and it shows API-key authentication through an x-api-key header for the documented examples.

That information is a baseline, not permission to guess the rest of a particular flow. The flow ID or endpoint name, compatible input type, useful output component, exposed tweak fields, authentication settings, and response path all depend on what is deployed. Langflow's API Access panel generates code for the selected flow. Keep that snippet or an equivalent reviewed fixture beside the consumer contract.

Also inspect the API documentation served by the test deployment. Langflow's official API guide points to the deployment's /docs page for its OpenAPI interface. This matters during upgrades because a client may target a server version different from the article or from a developer laptop. A contract test should report the deployment version and endpoint it exercised rather than claiming all Langflow installations behave identically.

Split the contract into layers. The transport layer covers base URL, path, TLS, authentication, content type, timeout, and status. The request layer covers required fields, allowed values, and the concrete flow's input expectations. The execution layer proves the selected flow ran with the intended configuration. The response layer verifies the exact value and location consumed by the application. Session and tweak behavior sit across those layers because they change execution without changing the basic route.

Keep the positive probe deterministic. A general-purpose chat flow backed by a live model is a poor request-schema oracle because model output can vary and provider availability can fail. Create or designate a small test flow whose output includes a fixed prefix and an escaped copy or deterministic transformation of approved input. If that is not possible, assert a stable structural field plus a flow-owned value that does not depend on model wording.

Do not search the entire JSON tree for the input text. Many responses include echoed inputs, trace metadata, or component state. A recursive search can pass even when the final output is empty. Configure the exact response location used by the application. If that path changes after a Langflow or flow update, the test should fail and force a consumer review.

The examples below use a JSON Pointer supplied by the test environment. That is an application choice, not a claim that Langflow guarantees one response shape for every flow and version. Obtain the pointer from the actual probe response, review it against the consumer, and keep it under version control or environment configuration.

Send a request that preserves failure evidence

This client uses the documented simplified endpoint and keeps non-2xx bodies available for assertions. It validates only application-side invariants that are safe to enforce before the network call. The server remains responsible for its own request validation.

from __future__ import annotations

from dataclasses import dataclass
import json
from typing import Any, Literal
from urllib.parse import quote

import requests


InputType = Literal["chat", "text", "any"]
OutputType = Literal["chat", "text", "any", "debug"]


@dataclass(frozen=True)
class RunInput:
    input_value: str
    input_type: InputType = "chat"
    output_type: OutputType = "chat"
    output_component: str | None = None
    session_id: str | None = None
    tweaks: dict[str, Any] | None = None

    def payload(self) -> dict[str, Any]:
        if not isinstance(self.input_value, str) or not self.input_value:
            raise ValueError("input_value must be a non-empty string")
        body: dict[str, Any] = {
            "input_value": self.input_value,
            "input_type": self.input_type,
            "output_type": self.output_type,
        }
        if self.output_component is not None:
            body["output_component"] = self.output_component
        if self.session_id is not None:
            body["session_id"] = self.session_id
        if self.tweaks is not None:
            body["tweaks"] = self.tweaks
        return body


class LangflowRunClient:
    def __init__(
        self,
        base_url: str,
        api_key: str,
        *,
        timeout_seconds: float = 30.0,
        session: requests.Session | None = None,
    ) -> None:
        if not base_url.startswith(("http://", "https://")):
            raise ValueError("base_url must include http:// or https://")
        self.base_url = base_url.rstrip("/")
        self.api_key = api_key
        self.timeout_seconds = timeout_seconds
        self.session = session or requests.Session()

    def run(self, flow_id_or_name: str, request: RunInput) -> requests.Response:
        if not flow_id_or_name:
            raise ValueError("flow_id_or_name is required")
        encoded_flow = quote(flow_id_or_name, safe="")
        return self.session.post(
            f"{self.base_url}/api/v1/run/{encoded_flow}",
            headers={
                "accept": "application/json",
                "Content-Type": "application/json",
                "x-api-key": self.api_key,
            },
            json=request.payload(),
            timeout=self.timeout_seconds,
        )


def resolve_json_pointer(document: Any, pointer: str) -> Any:
    if pointer == "":
        return document
    if not pointer.startswith("/"):
        raise ValueError("JSON pointer must be empty or start with /")
    current = document
    for raw_token in pointer[1:].split("/"):
        token = raw_token.replace("~1", "/").replace("~0", "~")
        if isinstance(current, list):
            current = current[int(token)]
        elif isinstance(current, dict):
            current = current[token]
        else:
            raise TypeError(f"cannot descend through {type(current).__name__}")
    return current

The client intentionally does not call raise_for_status() inside run. A positive helper can do that after saving the body, but negative contract tests need to inspect validation responses. Log the status, response content type, request correlation header if your deployment provides one, and a redacted body. Never print the API key.

Both literals are copied from the installed package rather than from a documentation page. On Langflow 1.7.2 they are declared in lfx/schema/schema.py, which is where SimplifiedAPIRequest picks them up, and python -c "from lfx.schema.schema import InputType, OutputType; print(InputType, OutputType)" prints them from the deployment you actually run. That distinction matters: the two sets are not symmetric, any is a valid input type as well as an output type, and text is a valid output type even though a reader who trusted an out-of-date list would find their own client rejecting it before the request left the process. Pin the literals to the server version your team supports, refresh them from the installed source during an upgrade, and treat a mismatch between the client type and /docs as a released defect rather than a documentation nuance.

Timeout choice is another application decision. The illustrative default in this client is not a measured recommendation. A model-backed flow can take longer, while a deterministic probe should normally finish quickly. Configure the value from the test environment and record timeout separately from server responses so CI does not report a client deadline as a Langflow validation error.

Prove failure at the request boundary and success at the output boundary

The first worked case belongs to the application client rather than the generic endpoint schema. Current Langflow documentation makes input_value optional with a default of null, so omission does not carry a universal 422 contract. A particular flow can still need non-empty input to do useful work. This example intentionally narrows RunInput to a non-empty string and tests that application-owned rule before making a request.

If you add an integration case that omits input_value, assert the status and output defined for the pinned deployment and the selected flow. Do not hard-code 422 from the endpoint schema alone. Validation payload wording and structure can change with server dependencies, so retain a redacted body and assert a stable machine-readable error field only when the deployed contract defines one.

The second case receives 200 but exposes the wrong consumer path. Perhaps the flow has multiple outputs, or an upgrade changes nesting. A broad assertion such as "CONTRACT_OK" in response.text can pass because a trace field echoes the prompt. Resolve the configured output pointer and compare the exact sentinel generated by the probe flow.

The third case leaks conversation state between tests. Langflow documentation describes session_id as the conversation context identifier and shows reusing it to continue a conversation. A suite that hard-codes test-session across parallel cases can make one test influence another. Generate a unique session ID for independent tests, then write a separate continuation scenario that deliberately reuses one ID.

The integration tests below require a dedicated deployment and probe flow. They skip when the environment is absent, which keeps local unit runs honest. In the protected integration job, missing variables should be checked before pytest so a misconfigured CI environment cannot turn all cases into skipped passes.

from __future__ import annotations

import json
import os
from uuid import uuid4

import pytest

from langflow_contract import LangflowRunClient, RunInput, resolve_json_pointer


REQUIRED = (
    "LANGFLOW_TEST_URL",
    "LANGFLOW_TEST_API_KEY",
    "LANGFLOW_TEST_FLOW_ID",
    "LANGFLOW_TEST_OUTPUT_POINTER",
)


@pytest.fixture(scope="session")
def contract_env() -> dict[str, str]:
    missing = [name for name in REQUIRED if not os.getenv(name)]
    if missing:
        pytest.skip(f"Langflow contract environment not configured: {missing}")
    return {name: os.environ[name] for name in REQUIRED}


@pytest.fixture(scope="session")
def client(contract_env: dict[str, str]) -> LangflowRunClient:
    return LangflowRunClient(
        contract_env["LANGFLOW_TEST_URL"],
        contract_env["LANGFLOW_TEST_API_KEY"],
        timeout_seconds=float(os.getenv("LANGFLOW_TEST_TIMEOUT_SECONDS", "30")),
    )


def test_application_client_requires_nonempty_input_value() -> None:
    with pytest.raises(ValueError, match="input_value must be a non-empty string"):
        RunInput(input_value="").payload()


def test_probe_value_reaches_the_configured_output(
    client: LangflowRunClient, contract_env: dict[str, str]
) -> None:
    nonce = uuid4().hex
    response = client.run(
        contract_env["LANGFLOW_TEST_FLOW_ID"],
        RunInput(
            input_value=f"probe:{nonce}",
            input_type="chat",
            output_type="chat",
            session_id=f"contract-{uuid4().hex}",
        ),
    )

    assert response.status_code == 200, response.text
    assert response.headers.get("content-type", "").split(";", 1)[0] == "application/json"
    actual = resolve_json_pointer(
        response.json(), contract_env["LANGFLOW_TEST_OUTPUT_POINTER"]
    )
    assert actual == f"CONTRACT_OK:{nonce}"


def test_two_independent_runs_use_different_sessions(
    client: LangflowRunClient, contract_env: dict[str, str]
) -> None:
    session_ids = [f"contract-{uuid4().hex}" for _ in range(2)]

    responses = [
        client.run(
            contract_env["LANGFLOW_TEST_FLOW_ID"],
            RunInput(input_value="probe:isolation", session_id=session_id),
        )
        for session_id in session_ids
    ]

    sent_sessions = [
        json.loads(response.request.body)["session_id"] for response in responses
    ]
    assert sent_sessions == session_ids
    assert [response.status_code for response in responses] == [200, 200]

Note what the last test asserts and, more importantly, what it stopped asserting. An earlier version compared session_ids[0] with session_ids[1] immediately after generating both from uuid4().hex. That comparison restates the two lines above it and no change to the client or the server could ever make it fail, which makes it a decoration rather than a check. The replacement reads the session identifier back out of the request body that requests actually prepared, so it fails if RunInput.payload() stops forwarding session_id, if a helper starts reusing one identifier for the whole module, or if the client rewrites the field on its way out.

Even so, this test proves a client property and a completed round trip, not state isolation. A stronger probe flow should return an approved session-scoped counter or previous-message marker at the configured pointer. Then assert that each fresh session begins at the flow's defined initial state. Add that check only after the flow owner exposes a deterministic value.

The positive test has an oracle that can fail when the flow runs the wrong branch. Changing the output pointer, removing the sentinel transform, selecting a different output component, or dropping the request input changes actual. The nonce also prevents a stale cached constant from passing. It is not a secret and should not contain customer data.

The first test asserts the narrower application-client rule and makes no claim that Langflow rejects an omitted input_value. If the selected flow requires input, add a flow-specific integration assertion based on that deployed flow's behavior and /docs, not on a universal endpoint-schema assumption.

Expand the request matrix from the caller's real data, not generic fuzz labels. A chat client may need empty-looking whitespace, line breaks, emoji, non-Latin scripts, quotes, and a message near its own accepted-size limit. The probe flow should return an exact, deterministic encoding so the test can tell preservation from truncation or normalization. Do not assume that every transformation is wrong. If the application intentionally trims outer whitespace or normalizes line endings, enforce that rule in the client contract and make the expected value explicit.

JSON types deserve direct boundary cases. Under the current endpoint documentation, null and omission are allowed schema values, so test their outcome against the selected flow's pinned contract. Send a number, object, and array with the low-level session and assert the validation behavior documented by the deployment. These tests reveal a server upgrade that begins coercing values the consumer never intended to send. Keep them separate from RunInput unit tests, which should prove that the application rejects invalid values before using a credential or network connection.

Flow names and identifiers need boundary coverage too. The client URL-encodes the path segment because a human-readable endpoint name can contain characters that do not belong raw in a URL path. Test the exact identifiers your deployment policy permits. Do not invent support for slashes inside a name merely because percent-encoding exists; path routing and server lookup still define what is valid.

Input and output type defaults can hide drift. If the application depends on chat, send it explicitly as the examples do. A server default changing would then not alter behavior silently. If the consumer intentionally relies on defaults, write a paired test with and without the optional field and compare the flow-owned output. That makes the dependency visible and gives an upgrade review concrete evidence.

Output-component selection should be tested only when the application sends it. Capture the component identifier from the deployed flow's generated example, request it explicitly, and assert the configured output pointer. A negative control should select another approved output and prove the value changes. Without that control, an implementation that ignores output_component could still pass because the default happens to match.

Keep an error taxonomy at the test harness boundary. At minimum distinguish client validation, connection failure, client timeout, authentication response, server schema validation, flow execution response, response parsing, and semantic mismatch. Preserve the original status and safe body excerpt. Collapsing every exception into Langflow run failed removes the evidence needed to choose between client code, deployment configuration, flow design, and external providers.

One more mutation protects against a deceptive probe. Replace the request nonce with a different value but hold the expected sentinel at the old value. The test must fail. Then keep the request fixed and point the resolver at a metadata field containing the input; that must also fail because the exact CONTRACT_OK: output is absent there. These changes demonstrate that the oracle verifies the execution output rather than mere input echo.

Test tweaks and sessions as behavior, not payload decoration

Langflow documents tweaks as temporary run-time overrides for component parameters. The quickstart explains that they affect one run and do not modify the underlying flow configuration. A useful contract test therefore needs three calls: baseline, tweaked, and baseline again. The middle output must show the approved override; the third must return to the original value.

This test requires a dedicated component parameter whose effect is deterministic and safe. For example, a probe component can prefix output with a configured label. The tweak targets the exact component identifier and parameter generated by API Access. Do not use a live model temperature as the oracle. A changed temperature does not guarantee changed text, so the test could fail or pass randomly.

Component identifiers can change when flows are copied, imported, or rebuilt. That is precisely why a contract test should observe the tweak's effect. Merely asserting that a request containing the expected dictionary got status 200 cannot detect a stale identifier if the deployment ignores or tolerates it. The response needs a sentinel tied to the overridden component.

Do not assert one universal server behavior for an unknown tweak. Depending on the deployed version and validation path, an invalid component or field may be rejected or may fail to affect output. The portable contract is that your approved tweak produces the expected behavior on your pinned flow, while the baseline remains unchanged afterward.

Session tests need the same discipline. One set of cases uses a fresh ID per test and expects the initial state. Another sequence reuses an ID and expects the flow's documented continuation behavior. A third case uses a different ID and proves it cannot see the first session's marker. If the flow has no memory component, a continuation assertion is meaningless and should not be copied from another project.

Parallel execution can reveal session-key bugs. Run two sessions with distinct markers and interleave requests. The results must remain associated with their own markers. Keep this in a controlled test deployment because concurrency may call paid providers and produce load. A deterministic local component is preferable for the contract layer.

Avoid using personally identifiable content as the marker. Random hexadecimal strings are enough. Save request IDs, session IDs, statuses, and the approved output field in failure artifacts. Redact headers and any unneeded response branches before uploading them from CI.

When a test fails, reproduce with curl before changing assertions. The command below uses the documented path and fields. --fail-with-body preserves an error response while returning a failing exit status, and --show-error keeps the transport message visible. The API key remains in an environment expansion rather than the command text committed to the repository.

The preamble deserves a moment because it is easy to get wrong. A bare test -n "$VAR" line only sets an exit status; without set -e the script carries on and curl runs against an empty URL, producing a transport error that looks like a Langflow outage. The identical lines in the CI job further down do work, because GitHub Actions invokes run blocks with bash -e. A snippet a reader pastes into a terminal has no such wrapper, so it needs the shebang, the shell options, and a message that says which variable is missing.

#!/usr/bin/env bash
set -euo pipefail

: "${LANGFLOW_TEST_URL:?Set LANGFLOW_TEST_URL, for example https://langflow.test.example}"
: "${LANGFLOW_TEST_API_KEY:?Set LANGFLOW_TEST_API_KEY from the test deployment}"
: "${LANGFLOW_TEST_FLOW_ID:?Set LANGFLOW_TEST_FLOW_ID for the probe flow}"

curl --silent --show-error --fail-with-body \
  --request POST \
  "$LANGFLOW_TEST_URL/api/v1/run/$LANGFLOW_TEST_FLOW_ID?stream=false" \
  --header "accept: application/json" \
  --header "Content-Type: application/json" \
  --header "x-api-key: $LANGFLOW_TEST_API_KEY" \
  --data '{
    "input_value": "probe:manual-diagnostic",
    "input_type": "chat",
    "output_type": "chat",
    "session_id": "contract-manual-diagnostic"
  }'

If this fails before an HTTP response, inspect DNS, TLS, proxy routing, and client timeout. If it returns an authentication response, compare the test deployment's authentication configuration and key scope. If it returns validation, compare the payload with API Access and /docs. If it returns 200 with the wrong value, move to the flow execution and output pointer. That order prevents a model prompt edit from masking a base URL typo.

Separate contract drift from nearby infrastructure failures

A reverse proxy can strip /api, rewrite the version segment, reject the body size, or remove x-api-key. The resulting status may resemble an application response. Inspect response headers and the body source before assigning the failure to Langflow. A proxy-branded HTML body is not a flow validation result, even if the numeric status matches one used by the application.

Authentication failure is not an input-shape failure. Use a valid, scoped test key for schema cases so the request reaches validation. Test missing or invalid credentials separately according to the deployment's documented security policy. Do not make broad assertions about one status code across every reverse proxy and Langflow configuration.

A flow can return 200 while an internal component produces a fallback value. The exact behavior belongs to the component and flow design. That is why the probe asserts a sentinel rather than generic success. For production flows, add contract checks for the fields the caller actually uses, and keep model-quality evaluation in a separate suite.

External model rate limits and outages can make a model-backed contract test fail for reasons unrelated to the run payload. A deterministic probe flow removes that dependency from the basic API gate. Scheduled end-to-end tests can cover the real model path and classify provider failures separately.

Response extraction can fail even when the flow output is correct. If the configured JSON Pointer no longer exists, save the redacted response and compare it with the deployed flow's API example. Do not change the pointer to the first path that makes CI green. Confirm that the application consumer has moved to the same field.

Flow identity drift deserves its own evidence. A human-readable endpoint name can be repointed, or a test environment can import a new flow under a different ID. Add an approved flow version or probe signature to the output if the platform and your flow design allow it. The response sentinel can include that version without relying on nondeterministic text.

Streaming changes the response contract. The examples above request non-streaming behavior and parse one JSON document. Do not reuse the same parser for a streaming response. Add a separate test that understands the deployment's documented event format, termination signal, and partial-error handling when your application enables streaming.

Health checks answer a different question. A healthy Langflow service can still host the wrong flow or reject your payload. A successful probe can coexist with a failing external model dependency that the probe does not use. Keep service health, trigger contract, and production workflow evaluation as separate signals.

Roll the checks into CI without turning flakiness into policy

Run payload builder and JSON Pointer unit tests on every pull request without a server. Run the deterministic probe against a dedicated Langflow deployment when protected credentials are available. Run model-backed production-like evaluations on a schedule or before a controlled release. This split keeps basic request regressions fast and makes external variability visible instead of normalizing retries.

Fail the integration job before pytest when required environment variables are missing. The fixture's local skip is convenient for developers, but a protected CI job should not report success after skipping every contract case. The shell checks below make that policy explicit, and they are load bearing here for a reason worth stating: Actions runs each run block through bash -e, so the first empty variable ends the step. Copy those same lines into an interactive shell without set -e and they check nothing.

name: langflow-run-contract

on:
  pull_request:
    paths:
      - "clients/langflow/**"
      - "tests/langflow_contract/**"
  workflow_dispatch:

jobs:
  run-contract:
    runs-on: ubuntu-latest
    environment: langflow-test
    timeout-minutes: 15
    env:
      LANGFLOW_TEST_URL: ${{ vars.LANGFLOW_TEST_URL }}
      LANGFLOW_TEST_FLOW_ID: ${{ vars.LANGFLOW_TEST_FLOW_ID }}
      LANGFLOW_TEST_OUTPUT_POINTER: ${{ vars.LANGFLOW_TEST_OUTPUT_POINTER }}
      LANGFLOW_TEST_API_KEY: ${{ secrets.LANGFLOW_TEST_API_KEY }}
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
          cache: "pip"
          cache-dependency-path: requirements-test.txt
      - run: python -m pip install -r requirements-test.txt
      - name: Require the protected integration environment
        run: |
          test -n "$LANGFLOW_TEST_URL"
          test -n "$LANGFLOW_TEST_FLOW_ID"
          test -n "$LANGFLOW_TEST_OUTPUT_POINTER"
          test -n "$LANGFLOW_TEST_API_KEY"
      - run: python -m pytest tests/langflow_contract -q

Declare cache-dependency-path alongside the cache key. Its default globs are **/requirements.txt and **/pyproject.toml, so a repository pinning its client suite in requirements-test.txt fails during setup and never reaches the payload contract at all.

The costs are concrete. A dedicated deployment needs maintenance. Probe flows can drift from application flows. Integration tests consume network time and may consume provider capacity if the flow is not deterministic. Exact output pointers couple tests to a response shape. Unique sessions leave test records that may need cleanup or retention limits.

Manage that cost with ownership and versioning, not weaker assertions. Store the flow identifier, expected sentinel format, output pointer, supported server version, and owning team together. Review them when the flow changes. Keep the generated API Access example or its essential fields available for comparison.

Do not snapshot an entire Langflow response by default. Metadata, timing, IDs, and nested component details can change without breaking the consumer. Assert the transport fields and output values that form the real contract, then attach a redacted full response only on failure. This reduces noisy updates while still catching meaningful drift.

Finally, do not aim these checks at production as a shortcut. A contract probe can create messages, consume quotas, trigger tools, or write external state depending on the flow. Use a test flow with harmless components and a test key. If a production smoke check is separately approved, give it a read-only path, a clear budget, and its own operational policy.

Know when the run endpoint is not the failing boundary

Do not rewrite the payload when the request never reaches Langflow. DNS errors, TLS failures, proxy HTML, and client-side timeouts need transport diagnosis. Preserve the error layer so a failed connection does not become “invalid input_value” in the ticket.

Avoid using the simplified trigger when the workflow requires a different documented API with a different contract. Langflow exposes other trigger and workflow endpoints. Select the endpoint intentionally and test its own schema. Do not wrap or rename fields until a 200 appears.

Skip live semantic assertions in pull-request CI when the output depends on an unconstrained model response. Test the payload with a deterministic probe, then evaluate semantic quality in a suite designed for model variability. Retrying a flaky assertion until it passes measures luck.

Do not use runtime tweaks as a configuration deployment system. The documentation describes them as temporary overrides for one run. Persistent flow changes belong in the flow lifecycle and should be reviewed as such.

Most importantly, do not call the contract green because the server returned JSON. The caller needs a particular value from a particular flow under a particular session and configuration. Assert that boundary, keep the failure evidence, and let unrelated layers fail under their own names.