Build a service for API_CALL
Goal
You will have an HTTP endpoint that the API_CALL formula function can call during a model calculation, and that returns a result Valsight turns into a node.
The examples use Python with FastAPI and pandas. Nothing about the contract is specific to either: it is one JSON POST in and one JSON body out, so any language and framework will do.
Prerequisites
- A Valsight account with administrator permissions
- Network access to your Valsight instance
Steps
1. Understand the contract
A model node calls your endpoint like this:
API_CALL("https://your-service/optimize", ['Volume', 'Capacity'], ["Country", "Site"])Valsight POSTs the values of
VolumeandCapacity, and expects a cube back on the levelsCountryandSite. Your service receives:Key Value inputsThe input nodes, keyed by position in the formula as a string: "0","1". Each value is a cube.result_levelsThe requested level names. A JSON array since 7.3; one string, for example "[Country, Site]", up to 7.2. See Read the requested levels.messageThe formula’s messageargument, passed through untouched.versionThe formula’s versionargument, passed through untouched.formatAlways "JSON".maximum_mbs_to_returnThe instance’s response size limit, as a string. A cube carries its columns as metadata and its values as rows:
{ "version": "1", "columnMetadata": [ {"name": "Country", "columnType": "level", "levelId": 118}, {"name": "Site", "columnType": "level", "levelId": 101}, {"name": "measure", "columnType": "measure"} ], "tableMetadata": {"tableType": "FACT", "tableName": "Input 0"}, "rowData": [["DE", "S-100", "1200"], ["DE", "S-101", "900"]] }You must answer with HTTP 200–299 and this envelope:
{ "response_status": "SUCCESS", "error_message": "", "output": { ... a cube in the same shape ... } }Valsight reads only those three keys. See API_CALL for the full reference.
Read the requested levels
result_levelschanged shape in 7.3: it is a JSON array from 7.3 onwards, and a single rendered string up to 7.2. Accept both and one service works against every release:def requested_levels(body): levels = body.get("result_levels") # 7.3 and later: a real JSON array if isinstance(levels, list): return [str(name) for name in levels] # up to 7.2: one rendered string, for example "[Country, Site]" if isinstance(levels, str) and levels.startswith("[") and levels.endswith("]"): inner = levels[1:-1].strip() return [part.strip() for part in inner.split(",")] if inner else [] return []On 7.2 and earlier this cannot be made reliable: a level name may itself contain a comma, and the string form gives you no way to tell that comma apart from the separator. The array form in 7.3 removes the ambiguity, which is why the key changed shape.
2. Convert a cube to a DataFrame and back
These two helpers are the whole translation layer. Everything else in your service is ordinary application code.
import pandas as pd def cube_to_frame(cube, measure_name): """Turn one Valsight cube into a DataFrame with a numeric measure column.""" columns = [column["name"] for column in cube["columnMetadata"]] frame = pd.DataFrame(cube["rowData"], columns=columns) frame = frame.rename(columns={"measure": measure_name}) frame[measure_name] = pd.to_numeric(frame[measure_name], errors="coerce") return frame def frame_to_cube(frame, level_names, measure_column, source_cube, table_name): """Turn a DataFrame back into the cube Valsight expects in `output`.""" by_name = {column["name"]: column for column in source_cube["columnMetadata"]} metadata = [ by_name.get(name, {"name": name, "columnType": "level"}) for name in level_names ] metadata.append({"name": "measure", "columnType": "measure"}) ordered = frame[level_names + [measure_column]] return { "version": 1, "columnMetadata": metadata, "tableMetadata": {"tableType": "FACT", "tableName": table_name}, "rowData": ordered.values.tolist(), }Two details in there earn their place:
pd.to_numericis not optional. Every value inrowDataarrives as a string, numbers included.frame_to_cubecopiescolumnMetadatafrom an input cube rather than writing its own. Those entries carry thelevelIdvalues Valsight uses to map values back onto its dimensions; an invented id fails the conversion.
On the way out you may return real JSON numbers. There is no need to turn your results back into strings.
3. Start with an echo endpoint
Wire up the simplest possible endpoint first. It proves the allow-list, the network path and the envelope are right, with no modelling questions attached.
from fastapi import FastAPI, Request app = FastAPI() @app.post("/echo") async def echo(request: Request): body = await request.json() cube = body.get("inputs", {}).get("0") if cube is None: return failure("No input cube received at position 0.") return {"response_status": "SUCCESS", "error_message": "", "output": cube}Call it from a model with
API_CALL("https://your-service/echo", ['Some Node'], ["Year"]). The node should show the input node’s own values.4. Report errors so the user can see them
Return HTTP 200 with
response_statusset toERROR, not a 4xx status:def failure(message): return { "response_status": "ERROR", "error_message": message, "output": None, }On a status outside 200–299 Valsight discards the body and shows a generic transport error, so your text never reaches the user. With
ERRORit is displayed on the node, which is where the person who wrote the formula will look.5. Handle the four cases that actually occur
These are the ones that turn up in real models rather than in tests.
Case What to do The input cube is empty Return an empty cube of the right shape, not an error. A modeller who filtered everything out in a scenario legitimately sends nothing, and the node should calculate to zero. The solver drops rows Return the dropped members at 0rather than leaving them out. A missing row reads as no value, so downstream nodes fall back to a previous figure instead of registering the change.A requested level is not in the cube Fail with a clear error_messagenaming the level. Returning a cube of the wrong shape is far harder to diagnose from inside a model.The cube carries levels you did not ask for Ignore them. Input nodes are sent with full dimensionality, so a node on Sitealso arrives with itsCountrycolumn. Read levels by name.6. Get the endpoint allow-listed
API_CALL refuses every URL until an administrator sets
featureFlags.apiCall.allowedUrlsRegexon the instance — a Java regular expression that your endpoint must match. Empty is the default and allows nothing.Two other instance limits are worth knowing before you design the payload:
Setting Default Meaning featureFlags.apiCall.maxRowsToSend150000Maximum rows across all input nodes combined, per request. featureFlags.apiCall.maxMbsToReturn24Maximum response size read, in megabytes. Requests can additionally be restricted to the instance’s outbound IP address. Ask your Valsight support team to set both up.
7. Keep it fast
The default
timeoutis 30 seconds, set per formula. Aim far below it: the node recalculates on every user interaction, so anything past roughly two seconds makes the model feel unresponsive long before the timeout matters.If the underlying computation is genuinely slow, that is a sign the work belongs in an uploaded data table via DATA rather than in a live call.
Verify
Test the endpoint before pointing a model at it. The request body is plain JSON, so no Valsight instance is needed:
curl -s localhost:8000/optimize \
-H 'Content-Type: application/json' -d '{
"format": "JSON",
"version": "1",
"message": "",
"maximum_mbs_to_return": "24",
"result_levels": "[Country, Site]",
"inputs": {
"0": {
"version": "1",
"columnMetadata": [
{"name": "Country", "columnType": "level", "levelId": 118},
{"name": "Site", "columnType": "level", "levelId": 101},
{"name": "measure", "columnType": "measure"}
],
"tableMetadata": {"tableType": "FACT", "tableName": "Input 0"},
"rowData": [["DE","S-100","1200"],["DE","S-101","900"]]
}
}
}'You are done when all four of these hold:
- The response is HTTP 200 and
response_statusisSUCCESS. output.columnMetadatanames exactly the levels you asked for, plusmeasure.- A node using the formula calculates and shows those values.
- Deliberately breaking the service — stopping it, or returning
response_status: "ERROR"— puts a readable message on the node rather than a generic failure.
The complete service
Everything above, assembled into one file that runs as it stands. Save it as
valsight_api_call_example.py, install fastapi, uvicorn and pandas, and start it
with uvicorn valsight_api_call_example:app --port 8000.
Only solve() is about the problem. Replace it with a call to your own service and
leave the rest — that is the Valsight-facing boilerplate this page has been building up.
"""
Valsight API_CALL — reference service (FastAPI)
==============================================
A Valsight model node can call an external HTTP service in the middle of a
calculation and use the response as a node:
API_CALL("https://your-service/optimize", ['Volume', 'Capacity'], ["Country", "Site"])
\\______________ ______________/ \\_________ ________/ \\________ _______/
\\/ \\/ \\/
endpoint (allow-listed) input nodes -> "inputs" levels you must
return
Valsight POSTs one JSON body, expects one JSON body back, and turns the
`output` cube into the node's value. That is the whole contract.
This file is a runnable reference implementation. Replace `solve()` with a call
to your own service — everything around it is the Valsight-facing boilerplate.
pip install "fastapi[standard]" pandas
uvicorn valsight_api_call_example:app --host 0.0.0.0 --port 8000
Verified against the Valsight product code (OlapApiCallCubeDefinition.groovy)
on 2026-08-28, release 7.2.
THE REQUEST BODY Valsight sends
-------------------------------
{
"inputs": { # one entry per input node, keyed by its
"0": {<cube>}, # position in the formula, as a STRING
"1": {<cube>}
},
"result_levels": ["Country", "Site"], # array from 7.3 on; up to 7.2
# the string "[Country, Site]"
"message": "free text", # the formula's `message` argument
"version": "1", # the formula's `version` argument
"format": "JSON",
"maximum_mbs_to_return": "24" # a string, from an instance setting
}
A cube looks like this. Note that **every value in `rowData` is a string**,
numbers included:
{
"version": "1",
"columnMetadata": [
{"name": "Country", "columnType": "level", "levelId": 118},
{"name": "Site", "columnType": "level", "levelId": 101},
{"name": "measure", "columnType": "measure"}
],
"tableMetadata": {"tableType": "FACT", "tableName": "Input 0"},
"rowData": [["DE", "S-100", "1200"], ["DE", "S-101", "900"]]
}
Three things that surprise people:
1. Values are strings. Always `pd.to_numeric(..., errors="coerce")`.
2. Cubes arrive with **full dimensionality**. If the input node is dimensioned
on Site only, but Site sits under Country, you still receive a Country
column. That is the only channel through which Valsight communicates
dimensionality, so read the levels you need by name and ignore the rest.
3. `result_levels` changed shape in 7.3, from a rendered string to a JSON
array. Up to 7.2 it cannot be parsed reliably, because a level name may
itself contain a comma. `requested_levels()` below reads either form.
THE RESPONSE BODY Valsight expects
----------------------------------
{
"response_status": "SUCCESS", # or "ERROR"
"error_message": "", # shown to the user when status is ERROR
"output": {<cube>} # same shape as an input cube
}
Valsight reads only `response_status`, `error_message` and `output`; any other
key is ignored. The HTTP status must be in 200-299, otherwise the calculation
fails before the body is even looked at.
On the way in every value is a string; on the way out you may return real JSON
numbers. No need to stringify what you computed.
`output` must carry exactly the levels that were requested, plus the `measure`
column. Reuse the input cube's `columnMetadata` for those levels rather than
writing your own — see `level_metadata()`.
LIMITS AND TIMEOUTS
-------------------
* Default timeout is 30 s, set per formula via the `timeout` argument. A node
is recalculated on every user interaction, so in practice aim well under
~2 s or the model stops feeling interactive.
* Max rows sent: 150 000 by default, counted as the **sum over all** input
cubes.
* Max response read: 24 MB by default.
* The endpoint must match the URL allow-list on the Valsight instance
(`featureFlags.apiCall.allowedUrlsRegex`) and can be locked to your source
IP. Both are set by a Valsight administrator.
"""
from __future__ import annotations
import logging
from typing import Any, Dict, List
import pandas as pd
from fastapi import APIRouter, FastAPI, Request
logger = logging.getLogger("valsight.api_call")
PAGE = APIRouter()
# Levels this service knows how to work with. Keep the interface generic: read
# levels by name, pass through what you do not need.
SITE_LEVEL = "Site"
MEASURE = "measure"
# ---------------------------------------------------------------------------
# Valsight cube <-> pandas
# ---------------------------------------------------------------------------
def cube_to_frame(cube: Dict[str, Any], measure_name: str) -> pd.DataFrame:
"""Turn one Valsight cube into a DataFrame with a numeric measure column."""
columns = [column["name"] for column in cube["columnMetadata"]]
frame = pd.DataFrame(cube["rowData"], columns=columns)
frame = frame.rename(columns={MEASURE: measure_name})
frame[measure_name] = pd.to_numeric(frame[measure_name], errors="coerce")
return frame
def level_metadata(cube: Dict[str, Any], level_names: List[str]) -> List[Dict[str, Any]]:
"""Reuse the input cube's column metadata for the levels we return.
Copying it is deliberate: the metadata carries `levelId`, which Valsight
uses to map values back onto its own dimensions. Inventing an id here would
fail the conversion, so we look each level up and only fall back to a
name-only entry when the input did not carry it.
"""
by_name = {column["name"]: column for column in cube["columnMetadata"]}
return [
by_name.get(name, {"name": name, "columnType": "level"})
for name in level_names
]
def frame_to_cube(
frame: pd.DataFrame,
level_names: List[str],
measure_column: str,
source_cube: Dict[str, Any],
table_name: str,
) -> Dict[str, Any]:
"""Turn a DataFrame back into the cube Valsight expects in `output`."""
metadata = level_metadata(source_cube, level_names)
metadata.append({"name": MEASURE, "columnType": "measure"})
ordered = frame[level_names + [measure_column]]
return {
"version": 1,
"columnMetadata": metadata,
"tableMetadata": {"tableType": "FACT", "tableName": table_name},
"rowData": ordered.values.tolist(),
}
def requested_levels(body: Dict[str, Any]) -> List[str]:
"""The levels Valsight wants back, from either payload form.
`result_levels` changed shape in 7.3: a JSON array from 7.3 onwards, a
single rendered string up to 7.2. Accepting both keeps one service working
against every release.
"""
levels = body.get("result_levels")
# 7.3 and later: a real JSON array
if isinstance(levels, list):
return [str(name) for name in levels]
# up to 7.2: one rendered string, for example "[Country, Site]"
if isinstance(levels, str) and levels.startswith("[") and levels.endswith("]"):
inner = levels[1:-1].strip()
return [part.strip() for part in inner.split(",")] if inner else []
return []
# ---------------------------------------------------------------------------
# Response envelope
# ---------------------------------------------------------------------------
def success(output: Dict[str, Any], function_name: str) -> Dict[str, Any]:
return {
"format": "JSON",
"version": "1",
"function_name": function_name,
"response_status": "SUCCESS",
"error_message": "",
"output": output,
}
def failure(message: str) -> Dict[str, Any]:
"""Report a business error.
Return HTTP 200 with status ERROR, not a 4xx: on a non-2xx Valsight
discards the body and shows a generic transport error, so `message` would
never reach the user. With status ERROR it is displayed on the node.
"""
logger.warning("returning ERROR to Valsight: %s", message)
return {
"format": "JSON",
"version": "1",
"response_status": "ERROR",
"error_message": message,
"output": None,
}
# ---------------------------------------------------------------------------
# Endpoint 1 — connectivity check
# ---------------------------------------------------------------------------
@PAGE.post("/echo")
async def echo(request: Request) -> Dict[str, Any]:
"""Return input cube 0 unchanged.
The first thing to wire up: it proves the allow-list, the network path and
the envelope are right, without any modelling questions attached.
API_CALL("https://your-service/echo", ['Volume'], ["Country", "Site"])
"""
body = await request.json()
cubes = body.get("inputs", {})
logger.info("/echo keys=%s cubes=%s", list(body), list(cubes))
cube = cubes.get("0")
if cube is None:
return failure("No input cube received at position 0.")
return success(cube, "echo")
# ---------------------------------------------------------------------------
# Endpoint 2 — the optimizer call
# ---------------------------------------------------------------------------
def solve(volume: pd.DataFrame, capacity: pd.DataFrame, integer: bool) -> pd.DataFrame:
""">>> REPLACE THIS with a call to your own service. <<<
Everything above and below is Valsight plumbing; this is the only part that
is actually about the problem.
Contract: take the input frames, return a frame with the same level columns
plus one numeric result column named `optimized`.
The placeholder below allocates volume up to a per-country capacity limit,
scaling proportionally when demand exceeds it. It exists so the file runs
as-is — it is not a model of anything.
"""
data = volume.merge(capacity, on="Country", how="left")
limit = data["capacity"].fillna(float("inf"))
demanded = data.groupby("Country")["volume"].transform("sum")
scale = (limit / demanded).clip(upper=1.0).fillna(1.0)
data["optimized"] = data["volume"] * scale
if integer:
data["optimized"] = data["optimized"].round()
return data
@PAGE.post("/optimize")
async def optimize(request: Request) -> Dict[str, Any]:
"""Two input cubes in, one optimized cube out.
API_CALL("https://your-service/optimize",
['Volume per Site', 'Capacity per Country'],
["Country", "Site"],
"JSON", "1", "continuous", 20)
\\________/ \\/
message timeout, seconds
The optional arguments are positionally nested in the formula grammar: to
pass `timeout` you must also pass `format`, `version` and `message`.
"""
body = await request.json()
cubes = body.get("inputs", {})
# `message` is passed through verbatim, which makes it a convenient mode
# switch — you can change solver behaviour from the formula without
# touching the API.
message = body.get("message") or ""
integer = "integer" in message.lower()
levels = requested_levels(body)
logger.info(
"/optimize cubes=%s levels=%s integer=%s", list(cubes), levels, integer
)
volume_cube = cubes.get("0")
capacity_cube = cubes.get("1")
if volume_cube is None or capacity_cube is None:
return failure(
"Expected two input nodes: volume per site, then capacity per country. "
f"Received {len(cubes)}."
)
try:
volume = cube_to_frame(volume_cube, "volume")
capacity = cube_to_frame(capacity_cube, "capacity")
except (KeyError, TypeError, ValueError) as exc:
logger.exception("could not parse input cubes")
return failure(f"Could not read the input cubes: {exc}")
# Levels are requested by name. Fail loudly here rather than silently
# returning a cube of the wrong shape, which is much harder to debug from
# inside a model.
missing = [level for level in levels if level not in volume.columns]
if missing:
return failure(
f"Input cube 0 has no column(s) {missing}. "
f"It carries {sorted(volume.columns)}."
)
# An empty input is normal, not an error: a modeller who closed every site
# in a scenario sends nothing. Return an empty cube of the right shape so
# the node calculates to zero instead of failing.
if volume.empty:
logger.info("no rows to optimize, returning an empty cube")
empty = pd.DataFrame(columns=levels + ["optimized"])
return success(
frame_to_cube(empty, levels, "optimized", volume_cube, "Optimized Volume"),
"optimize",
)
result = solve(volume, capacity, integer)
# Sites the solver dropped must still appear, at zero. Leaving them out
# makes them missing rather than closed, and downstream nodes then read the
# previous value instead of nothing.
if SITE_LEVEL in levels:
result = (
volume[levels]
.merge(result[levels + ["optimized"]], on=levels, how="left")
.fillna({"optimized": 0.0})
)
output = frame_to_cube(
result, levels, "optimized", volume_cube, "Optimized Volume"
)
logger.info("/optimize returning %s rows", len(output["rowData"]))
return success(output, "optimize")
# ---------------------------------------------------------------------------
app = FastAPI(
title="Valsight API_CALL reference service",
description=__doc__,
version="1.0.0",
)
app.include_router(PAGE, prefix="/functions")
@app.get("/health")
async def health() -> Dict[str, str]:
return {"status": "ok"}
if __name__ == "__main__":
import uvicorn
logging.basicConfig(level=logging.INFO, format="[%(levelname)s] %(name)s: %(message)s")
uvicorn.run(app, host="0.0.0.0", port=8000)
# ---------------------------------------------------------------------------
# Try it without Valsight
# ---------------------------------------------------------------------------
#
# curl -s localhost:8000/functions/optimize \
# -H 'Content-Type: application/json' -d '{
# "format": "JSON",
# "version": "1",
# "message": "continuous",
# "maximum_mbs_to_return": "24",
# "result_levels": ["Country", "Site"],
# "inputs": {
# "0": {
# "version": "1",
# "columnMetadata": [
# {"name": "Country", "columnType": "level", "levelId": 118},
# {"name": "Site", "columnType": "level", "levelId": 101},
# {"name": "measure", "columnType": "measure"}
# ],
# "tableMetadata": {"tableType": "FACT", "tableName": "Input 0"},
# "rowData": [["DE","S-100","1200"],["DE","S-101","900"],["JP","S-200","400"]]
# },
# "1": {
# "version": "1",
# "columnMetadata": [
# {"name": "Country", "columnType": "level", "levelId": 118},
# {"name": "measure", "columnType": "measure"}
# ],
# "tableMetadata": {"tableType": "FACT", "tableName": "Input 1"},
# "rowData": [["DE","1500"],["JP","800"]]
# }
# }
# }' | python3 -m json.tool
#
# DE is capped: 1200 + 900 = 2100 demanded against 1500 available, so both
# sites scale by 1500/2100. JP stays at 400, well under its 800.Related
| Page | What it covers |
|---|---|
| API_CALL | The formula function itself: syntax, parameters, request and response reference, error messages. |
| DATA | Referencing data that is already uploaded, instead of calling a live service. |
| Introduction to Valsight API | Reading data out of Valsight, rather than being called by it. |