Example 04: Weather Advisor

The workflow

Pipeline structure

AI handles the parts the API returns as opaque strings; deterministic ops compute the flags and bands; a custom packing op bundles everything before the final AI call.

HTTPGetOp

wttr.in JSON

→

JSONExtractOp ×4

string fields

→

AIParseNumber ×3

AIClassifyMultiLabel

→

band + flags

det thresholds

→

PackOutfitOp

assemble prompt

→

AIComputeOp

+ probe → concat

det op

Deterministic steps

HTTPGetOp + JSONExtractOp (×4)

Fetches the wttr.in j1 format and pulls the four relevant string fields with dot-path selectors. All four extractions run in parallel once the HTTP body arrives.

ContextValOp (cold/mild/hot) + predicate gates + CoalesceNStringOp

Three ContextValOp vertices each read a band label from context and carry a numeric predicate. Exactly one fires; CoalesceNStringOp collapses the three wires into one band wire. No AI — just threshold comparisons on the parsed temp.

IfFloatGtOp (×2) — wet, windy

Compares precip_mm > 0.1 and wind_kph > 25 against constant wires. Returns a bool wire. No AI — the thresholds are your policy, expressed as data.

PackOutfitInputsOp

Custom op that accepts five input wires (band, wet, windy, temp_c, conditions) and formats them into: "Temperature: 10.0°C (mild), precipitation: dry, wind: calm, conditions: cloud". Keeps prompt construction out of the AI op config.

SelectStringOp + StringConcatOp

If the unusual-weather probe is true, selects a warning suffix; otherwise selects an empty string. Appends to the outfit advice. Same optional-suffix pattern as example 03.

AI op

AI steps

AIParseNumberOp (×3) — temp, precip, wind

The wttr.in API returns numbers as strings in the JSON (e.g. "10"). AIParseNumberOp converts each to a float64, handling any unit suffix or format variation. All three run concurrently.

AIClassifyMultiLabelOp — conditions

Classifies the weather description string into zero or more of: rain, snow, fog, sun, cloud, storm. Returns []string. One AI call; multiple labels can be present simultaneously.

AIComputeStringToStringOp — outfit advice

Reads the packed description string and writes exactly two sentences recommending specific clothing items. Runs after all deterministic signals are assembled.

AIBoolOp — unusual_flag (orthogonal probe)

Independently asks "is this weather unusual or extreme?" of the raw description string. Runs in parallel with the outfit advice op — neither blocks the other. Its output feeds the warning selection, not the main advice path.

Implementation

Walkthrough

1. JSON extraction + parallel number parsing

The wttr.in response stores numeric fields as JSON strings. JSONExtractOp pulls them out as Go strings; AIParseNumberOp converts each to a float64. The four extractions share the body wire and run concurrently, as do the three number parsers.

          
          
          
        
main.go — Stages 2 & 3
// Stage 2: extract four string fields in parallel.
for _, f := range []struct{ path, wire string }{
    {"current_condition.0.temp_C",        "temp_str"},
    {"current_condition.0.precipMM",      "precip_str"},
    {"current_condition.0.windspeedKmph", "wind_str"},
    {"current_condition.0.weatherDesc.0.value", "desc_str"},
} {
    b.Vertex("extract_"+f.wire).Op("JSONExtractOp").
        Input("JSON", "body").
        Input("Path", f.wire+"_path").
        Output("Value", f.wire)
}

// Stage 3: three AIParseNumberOps run concurrently.
b.Vertex("parse_temp").Op("AIParseNumberOp").
    Params(map[string]string{"operation": "extract temperature as plain number"}).
    Input("Input", "temp_str").
    Output("Result", "temp_c").
main.py — Stages 2 & 3
# Stage 2: extract four string fields in parallel.
for path, wire in [
    ("current_condition.0.temp_C", "temp_str"),
    ("current_condition.0.precipMM", "precip_str"),
    ("current_condition.0.windspeedKmph", "wind_str"),
    ("current_condition.0.weatherDesc.0.value", "desc_str"),
]:
    b.vertex(f"extract_{wire}").op("JSONExtractOp").input("JSON", "body").input("Path", f"{wire}_path").output("Value", wire)

# Stage 3: three AIParseNumberOps run concurrently.
b.vertex("parse_temp").op("AIParseNumberOp").params({"operation": "extract temperature as plain number"}).input("Input", "temp_str").output("Result", "temp_c")
index.ts — Stages 2 & 3
// Extracts strings deterministically
const tempStr = wf.op({ body }, ({ body }) => ops.json.jsonExtract(body, PATH_TEMP), { name: "extract_temp" });
const precipStr = wf.op({ body }, ({ body }) => ops.json.jsonExtract(body, PATH_PRECIP), { name: "extract_precip" });
const windStr = wf.op({ body }, ({ body }) => ops.json.jsonExtract(body, PATH_WIND), { name: "extract_wind" });
const descStr = wf.op({ body }, ({ body }) => ops.json.jsonExtract(body, PATH_DESC), { name: "extract_desc" });

// AI parses strings into floats concurrently
const tempC = wf.ai.parseNumber(tempStr, { operation: "extract temperature", name: "parse_temp" });
const precipMM = wf.ai.parseNumber(precipStr, { operation: "extract precipitation", name: "parse_precip" });
const windKph = wf.ai.parseNumber(windStr, { operation: "extract wind speed", name: "parse_wind" });

2. Deterministic temperature band with gated constants

Three ContextValOp vertices each read a band label from context and carry a numeric predicate. The predicates compare temp_c against thresholds. Exactly one fires per execution; CoalesceNStringOp merges the three wires into a single band string. Zero AI tokens for this categorisation.

          
          
          
        
main.go — Stage 4 (temperature band)
for _, band := range []struct{ name, cond string }{
    {"cold", "temp_is_cold"},  // temp < 10°C
    {"mild", "temp_is_mild"},  // 10 ≤ temp < 22°C
    {"hot",  "temp_is_hot"},   // temp ≥ 22°C
} {
    // Registered: builtin.ContextValFactory[string](coldBandKey{} / mildBandKey{} / hotBandKey{})
    // Injected:   ctx = context.WithValue(ctx, coldBandKey{}, "cold") etc.
    b.Vertex(band.name+"_const").Op(band.name+"_const").
        Condition(band.cond).
        ConditionInput("temp_c").
        Output("Result", band.name+"_band")
}
b.Vertex("band_merge").Op("CoalesceNStringOp").
    Params(map[string]int{"n": 3}).
    Merge(config.MergeCoalesce).
    Input("Input0", "cold_band").
    Input("Input1", "mild_band").
    Input("Input2", "hot_band").
    Output("Result", "band")
main.py — Stage 4 (temperature band)
for name, cond in [
    ("cold", lambda i: i.get("temp_c") < 10.0),
    ("mild", lambda i: 10.0 <= i.get("temp_c") < 22.0),
    ("hot", lambda i: i.get("temp_c") >= 22.0),
]:
    b.register_predicate(f"temp_is_{name}", cond)
    b.vertex(f"{name}_const").op("ContextValOp").params({"key": f"{name}_band_val"}).condition(f"temp_is_{name}").output("Result", f"{name}_band")

b.vertex("band_merge").op("CoalesceNStringOp").params({"n": 3}).input("Input0", "cold_band").input("Input1", "mild_band").input("Input2", "hot_band").output("Result", "band")
index.ts — Stage 4 (temperature band)
const cold = wf.op({ tempC }, () => "cold", { name: "cold", condition: ({ tempC }) => tempC < 10.0 });
const mild = wf.op({ tempC }, () => "mild", { name: "mild", condition: ({ tempC }) => tempC >= 10.0 && tempC < 22.0 });
const hot = wf.op({ tempC }, () => "hot", { name: "hot", condition: ({ tempC }) => tempC >= 22.0 });
const band = wf.coalesce([cold, mild, hot], { name: "band" });

3. Threshold-based boolean flags

IfFloatGtOp compares two float64 wires and emits a boolean. The thresholds (0.1 mm precip, 25 km/h wind) are constant wires — they can be changed by editing the graph, not the library code.

          
          
          
        
main.go — Stage 5 (wet / windy flags)
b.Vertex("precip_thresh").Op("ConstFloat64Op").
    Params(map[string]string{"Value": "0.1"}).
    Output("Result", "precip_thresh")
b.Vertex("wet_check").Op("IfFloatGtOp").
    Input("A", "precip_mm").
    Input("B", "precip_thresh").
    Output("Match", "wet")

b.Vertex("wind_thresh").Op("ConstFloat64Op").
    Params(map[string]string{"Value": "25.0"}).
    Output("Result", "wind_thresh")
b.Vertex("windy_check").Op("IfFloatGtOp").
    Input("A", "wind_kph").
    Input("B", "wind_thresh").
    Output("Match", "windy")
main.py — Stage 5 (wet / windy flags)
b.vertex("precip_thresh").op("ConstFloat64Op").params({"Value": 0.1}).output("Result", "precip_thresh")
b.vertex("wet_check").op("IfFloatGtOp").input("A", "precip_mm").input("B", "precip_thresh").output("Match", "wet")

b.vertex("wind_thresh").op("ConstFloat64Op").params({"Value": 25.0}).output("Result", "wind_thresh")
b.vertex("windy_check").op("IfFloatGtOp").input("A", "wind_kph").input("B", "wind_thresh").output("Match", "windy")
index.ts — Stage 5 (wet / windy flags)
const wet = wf.op({ precipMM }, ({ precipMM }) => precipMM > PRECIP_THRESH, { name: "wet_check" });
const windy = wf.op({ windKph }, ({ windKph }) => windKph > WIND_THRESH, { name: "windy_check" });

4. The orthogonal probe and warning suffix

The unusual-weather probe reads from desc_str — the same raw description string used earlier — without depending on any of the downstream signal wires. This means it can run in parallel with the number parsers. Its boolean output feeds a SelectStringOp that chooses between a warning string and an empty string; StringConcatOp appends the result to the outfit advice.

          
          
          
        
main.go — Stage 9 (orthogonal probe)
// Runs in parallel with parse_temp/parse_precip/parse_wind and classify.
b.Vertex("unusual_check").Op("AIBoolOp").
    Params(map[string]string{"predicate": "Is the described weather unusual or extreme?"}).
    Input("Input", "desc_str").
    Output("Result", "unusual_flag")

b.Vertex("warning_select").Op("SelectStringOp").
    Input("Cond", "unusual_flag").
    Input("IfTrue", "warning_str").  // "  ⚠ unusual weather"
    Input("IfFalse", "empty_str").
    Output("Result", "warning_suffix")

b.Vertex("final_concat").Op("StringConcatOp").
    Input("A", "outfit_advice").
    Input("B", "warning_suffix").
    Output("Result", "final_advice")
main.py — Stage 9 (orthogonal probe)
# Runs in parallel with parse_temp/parse_precip/parse_wind and classify.
b.vertex("unusual_check").op("AIBoolOp").params({"predicate": "Is the described weather unusual or extreme?"}).input("Input", "desc_str").output("Result", "unusual_flag")

b.vertex("warning_select").op("SelectStringOp").input("Cond", "unusual_flag").input("IfTrue", "warning_str").input("IfFalse", "empty_str").output("Result", "warning_suffix")

b.vertex("final_concat").op("StringConcatOp").input("A", "outfit_advice").input("B", "warning_suffix").output("Result", "final_advice")
index.ts — Stage 9 (orthogonal probe)
const unusual = wf.ai.bool(descStr, { predicate: "Is this unusual weather?", name: "unusual_check" });
const finalAdvice = wf.op(
  { outfitAdvice, unusual },
  ({ outfitAdvice, unusual }) => outfitAdvice + (unusual ? WARNING : ""),
  { name: "final_concat" },
);

Run it

Three fixtures cover all three temperature bands — London (mild), Reykjavik (cold, likely unusual), Singapore (hot).

          
          
          
        
shell
export CLAUDE_API_KEY=<your key>

# Live API
go run ./examples/weather-advisor --city London
go run ./examples/weather-advisor --city Reykjavik
go run ./examples/weather-advisor --city Singapore

# Offline fixtures
go run ./examples/weather-advisor --fixture examples/weather-advisor/testdata/weather/london.json
go run ./examples/weather-advisor --fixture examples/weather-advisor/testdata/weather/reykjavik.json
shell
export CLAUDE_API_KEY=<your key>

# Live API
python examples/weather_advisor/main.py --city London
python examples/weather_advisor/main.py --city Reykjavik
python examples/weather_advisor/main.py --city Singapore

# Offline fixtures
python examples/weather_advisor/main.py --fixture examples/weather_advisor/testdata/weather/london.json
python examples/weather_advisor/main.py --fixture examples/weather_advisor/testdata/weather/reykjavik.json
shell
export CLAUDE_API_KEY=<your key>

# Live API
npx tsx examples/weather_advisor/index.ts --city London
npx tsx examples/weather_advisor/index.ts --city Reykjavik
npx tsx examples/weather_advisor/index.ts --city Singapore

# Offline fixtures
npx tsx examples/weather_advisor/index.ts --fixture examples/weather_advisor/testdata/weather/london.json
npx tsx examples/weather_advisor/index.ts --fixture examples/weather_advisor/testdata/weather/reykjavik.json

Run it as an MCP server

The same binary is dual-mode. Pass --mcp and instead of running once and exiting it speaks the Model Context Protocol over stdin/stdout, exposing the entire workflow as a single MCP tool, advise_outfit. The Sparsi SDK derives the tool's input schema from the workflow's UserInput struct and validates every tools/call request against it, so an agent runs the whole fetch → bands/flags → classify → recommend pipeline with one tool call.

          
          
          
        
shell
# Speak MCP over stdin/stdout instead of running once
go run ./examples/weather-advisor --mcp
shell
# Speak MCP over stdin/stdout instead of running once
python examples/weather_advisor/main.py --mcp
shell
# Speak MCP over stdin/stdout instead of running once
npx tsx examples/weather_advisor/index.ts --mcp

Register it with any MCP client by pointing the client at that command:

.mcp.json

{
  "mcpServers": {
    "weather-advisor": {
      "command": "go",
      "args": ["run", "./examples/weather-advisor", "--mcp"]
    }
  }
}

Token usage

Measured on testdata/weather/london.json. The weather payload is tiny, making this the lowest-cost example. Four AIComputeOp calls cover the three number parsers and the final outfit writer.

Op	Calls	Input tokens	Output tokens	Total tokens
AIBoolOp	1	61	4	65
AIClassifyMultiLabelOp	1	84	4	88
AIComputeOp	4	303	77	380
Total	6	448	85	533

← Example 03 Next: HN Topic Brief →

Weather-Aware Outfit Advisor