Calling a Decision Model from Flink SQL with a UDF

Posted on:November 12, 2026

I’ve recently started a new role in Confluent Intelligence on the Stream Processing and Analytics Team. Our goal is to bring AI and ML models to real-time data streams. We support a number of models out of the box, including models for forecasting, anomaly detection, and sentiment classification. We also support Large Language Models through the AI_COMPLETE API.

But there are plenty of useful AI APIs that don’t speak the chat-completions protocol. This post shows how to call one of them, TypeSafe’s Jev “decisions” model served by OpenRouter, from Confluent Cloud Flink SQL with a REST connection and a small Java UDF. Jev is a model from TypeSafe AI that doesn’t generate text. You give it some context and a question, and it returns a decision with a probability in a few hundred milliseconds.

The Use Case

Jev was announced in September and answers yes/no questions about a piece of text and returns a score using “System One Thinking”. TypeSafe’s example is an agent safety check: given a proposed tool call, is it safe to run without a human approving it?

Every proposed action lands in a Kafka topic, and a Flink query scores each one as it arrives. Anything below a threshold goes to a human.

A User Defined Function

Jev isn’t a chat model. Using the OpenRouter API, it’s reachable at https://openrouter.ai/api/alpha/decisions and takes its own request shape:

{
  "model": "typesafe/jev-1.14",
  "state": "Task: clean up inactive accounts. Proposed tool call: delete_rows(...)",
  "questions": {
    "answer": {
      "type": "noul",
      "instructions": "Is this action safe to run without a human approving it first?",
      "criteria": {
        "true": "Reversible or low-impact, and clearly within the stated task.",
        "false": "Destructive, irreversible, or broader than the task requires."
      }
    }
  }
}

It returns a number rather than text:

{ "answers": { "answer": { "type": "noul", "noul": 0.07 } } }

The true and false entries in criteria describe what each answer means for your question. They aren’t returned. What comes back is the probability that the answer is yes, so a low score means the answer is probably no. You choose the threshold.

CREATE MODEL with the openai provider only sends chat-completions requests, so it can’t produce that body. A User Defined Function (UDF) is custom code that can send whatever format it likes, as long as it reaches the network through a connection. The connection contains the endpoint and the secret.

Overview:

  1. Store the Jev URL and an OpenRouter key in a REST connection.
  2. Write a scalar UDF that reads them, builds the request, and returns the score.
  3. Call it from SQL like any other function.

Step 1: Create the Connection

export OPENROUTER_API_KEY=...   # in your own shell

confluent flink connection create jev-decisions \
  --cloud aws --region us-west-2 \
  --type rest \
  --endpoint https://openrouter.ai/api/alpha/decisions \
  --token "$OPENROUTER_API_KEY"

Connection names allow only lowercase letters, digits, and hyphens.

Step 2: Write the UDF

The whole function uses only the JDK’s built-in HTTP client, so there are no extra dependencies.

The key idea is in open(). When you list a connection in USING CONNECTIONS, Flink hands its settings to the function as job parameters named <connection-name>.endpoint and <connection-name>.token.

package com.example.jev;

import java.io.IOException;
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.time.Duration;
import java.util.logging.Level;
import java.util.logging.Logger;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.apache.flink.table.functions.FunctionContext;
import org.apache.flink.table.functions.ScalarFunction;

public class JevDecisions extends ScalarFunction {

  static final String CONNECTION = "jev_decisions";
  static final String MODEL = "typesafe/jev-1.14";

  static final String DEFAULT_INSTRUCTIONS =
      "Is this action safe to run without a human approving it first?";
  static final String DEFAULT_WHEN_TRUE =
      "Reversible or low-impact, and clearly within the stated task.";
  static final String DEFAULT_WHEN_FALSE =
      "Destructive, irreversible, or broader than the task requires.";

  private static final Logger LOG = Logger.getLogger(JevDecisions.class.getName());
  private static final Pattern SCORE =
      Pattern.compile("\"noul\"\\s*:\\s*(-?[0-9.]+(?:[eE][-+]?[0-9]+)?)");

  private transient HttpClient client;
  private transient String endpoint;
  private transient String token;

  @Override
  public void open(FunctionContext context) {
    configure(param(context, "endpoint"), param(context, "token"));
  }

  // Connection names use hyphens; look under both spellings to be safe.
  private static String param(FunctionContext context, String property) {
    String value = context.getJobParameter(CONNECTION + "." + property, null);
    if (value == null) {
      value = context.getJobParameter(CONNECTION.replace('_', '-') + "." + property, null);
    }
    return value;
  }

  // Split out of open() so tests can configure the function without a Flink runtime.
  void configure(String endpoint, String token) {
    if (endpoint == null || token == null) {
      throw new IllegalStateException(
          "Missing connection settings. Create a REST connection named '"
              + CONNECTION.replace('_', '-')
              + "' and list it in USING CONNECTIONS.");
    }
    this.endpoint = endpoint;
    this.token = token;
    this.client = HttpClient.newBuilder().connectTimeout(Duration.ofSeconds(5)).build();
  }

  /** Scores the default question: "is this action safe to run without human approval?" */
  public Double eval(String state) {
    return eval(state, DEFAULT_INSTRUCTIONS, DEFAULT_WHEN_TRUE, DEFAULT_WHEN_FALSE);
  }

  /** Scores any yes/no question. Returns Jev's score, or NULL if the call failed. */
  public Double eval(String state, String instructions, String whenTrue, String whenFalse) {
    if (state == null || instructions == null) {
      return null;
    }
    try {
      HttpRequest request =
          HttpRequest.newBuilder(URI.create(endpoint))
              .timeout(Duration.ofSeconds(15))
              .header("Authorization", "Bearer " + token)
              .header("Content-Type", "application/json")
              .POST(HttpRequest.BodyPublishers.ofString(
                  buildRequest(state, instructions, whenTrue, whenFalse)))
              .build();
      HttpResponse<String> response = client.send(request, HttpResponse.BodyHandlers.ofString());
      if (response.statusCode() != 200) {
        LOG.warning("Jev call failed with HTTP " + response.statusCode() + ": " + response.body());
        return null;
      }
      return parseScore(response.body());
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
      return null;
    } catch (IOException | RuntimeException e) {
      LOG.log(Level.WARNING, "Jev call failed", e);
      return null;
    }
  }

  static String buildRequest(String state, String instructions, String whenTrue, String whenFalse) {
    return "{\"model\":" + quote(MODEL)
        + ",\"state\":" + quote(state)
        + ",\"questions\":{\"answer\":{\"type\":\"noul\""
        + ",\"instructions\":" + quote(instructions)
        + ",\"criteria\":{\"true\":" + quote(whenTrue == null ? "" : whenTrue)
        + ",\"false\":" + quote(whenFalse == null ? "" : whenFalse) + "}}}}";
  }

  static Double parseScore(String json) {
    Matcher m = SCORE.matcher(json);
    return m.find() ? Double.valueOf(m.group(1)) : null;
  }

  static String quote(String s) {
    StringBuilder sb = new StringBuilder("\"");
    for (int i = 0; i < s.length(); i++) {
      char c = s.charAt(i);
      switch (c) {
        case '"': sb.append("\\\""); break;
        case '\\': sb.append("\\\\"); break;
        case '\n': sb.append("\\n"); break;
        case '\r': sb.append("\\r"); break;
        case '\t': sb.append("\\t"); break;
        default:
          if (c < 0x20) {
            sb.append(String.format("\\u%04x", (int) c));
          } else {
            sb.append(c);
          }
      }
    }
    return sb.append('"').toString();
  }
}

Notes:

It never throws on a bad call. If an HTTP request fails and the UDF throws, Flink treats it as a job failure, restarts the statement, and retries the same row. If the problem is permanent, that loops forever. Returning NULL instead means one bad row costs you one NULL, and you can filter or alert on them downstream. The one exception is open(): if the connection isn’t wired up, failing loudly at startup is what you want.

It parses the score with a regex. A bit of a hack. It keeps the JAR free of a JSON library dependency. If the API grows more fields, switch to a real parser.

You can ask your own question. The class has two eval methods, and Flink picks one based on how many arguments you pass from SQL:

Step 3: Build it

Our example uses a standard pom.xml:

mvn clean package

Step 4: Upload and Register

confluent flink artifact create jev_udf \
  --artifact-file target/jev-flink-udf-1.0.jar \
  --cloud aws --region us-west-2 \
  --environment env-xxxxxx

That prints an artifact ID like cfa-xxxxxx. Use it in the USING JAR clause below.

CREATE FUNCTION jev_decide
  AS 'com.example.jev.JevDecisions'
  USING JAR 'confluent-artifact://cfa-xxxxxx'
  USING CONNECTIONS (`jev-decisions`);

Put the connection name in backticks, not single quotes.

Step 5: Call it

SELECT jev_decide(
  'Task: clean up inactive accounts. Proposed tool call: delete_rows(table="customers", where="last_login < 2023-01-01"). Context: 48,210 rows, no backup taken today.'
) AS score;

It returns 0.07, the same number curl gets from the API directly. A low score here means “not safe to run without approval”, which is the right call for a destructive delete with no backup.

If you forget USING CONNECTIONS, the function fails at startup with the “Missing connection settings” message, because the connection is what supplies the endpoint and key.

Streaming it Over Real Data

Confluent Cloud ships a read-only examples.marketplace catalog. The 4-argument form lets you ask your own question of its clicks stream, here: does this user agent look like an automated client?

SELECT click_id, user_agent,
  jev_decide(
    user_agent,
    'Does this user agent look like a human using a normal web browser?',
    'A mainstream desktop or mobile browser.',
    'A script, crawler, or automated client.'
  ) AS human_score
FROM examples.marketplace.clicks;

The query streams, and each row picks up a score as it goes through. In my run, “compatible” style agent strings scored around 0.95–0.99 and ordinary browsers around 0.13–0.29. This is sample data, so that shows the integration works, not that Jev is good at bot detection.

A yes/no question (Jev’s noul type) returns a probability rather than a label, so you pick the threshold in SQL:

Jev Decide

Notes

If you would like native support for the decision API in Flink, please reach out and let us know.