Multi-Step Requests
Merging Requests
In some cases, you may need to make multiple requests to different endpoints and merge their results. This example demonstrates how to combine data from two related API endpoints using a common identifier.
Scenario: We have two API endpoints for vehicle data:
/api/recalls- Contains vehicle recall information with avehicle_idfield/api/vehicles- Contains vehicle details with anidfield
We want to fetch the recall data for each vehicle and merge the results into a single output, where the recall record also includes the vehicle model.
Vehicles Payload: /api/vehicles
[
{
"id": 1,
"make": "Ford",
"model": "Explorer",
"year": 2021,
},
{
"id": 2,
"make": "Ford",
"model": "F-150",
"year": 2019,
},
{
"id": 3,
"make": "Ford",
"model": "Escape",
"year": 2020,
},
]
Recalls Payload: /api/recalls
[
{
"recall_id": "RCL-2026-0001",
"vehicle_id": 1,
"severity": "CRITICAL",
"description": "A-pillar malfunction may reduce structural integrity in a collision.",
"issued_date": "2026-01-14",
"status": "OPEN",
},
{
"recall_id": "RCL-2025-0187",
"vehicle_id": 2,
"severity": "HIGH",
"description": "Brake line corrosion may cause brake fluid leak and reduced braking performance.",
"issued_date": "2025-11-02",
"status": "OPEN",
},
{
"recall_id": "RCL-2024-0440",
"vehicle_id": 3,
"severity": "MEDIUM",
"description": "Airbag sensor calibration may incorrectly disable passenger airbag.",
"issued_date": "2024-06-18",
"status": "RESOLVED",
},
]
The snippet argument first gets the list of vehicle recalls from one endpoint and
stores it in the metadata. Then it makes a second request to another endpoint to get the list of
vehicles. Using the ID common to both lists, the results are merged into one list using the custom
step process_vehicle_recalls , defined below. The final result includes the recall data with
the vehicle models appended.
Snippet Argument
low_code:
version: 2
steps:
- http:
uri: "/api/recalls"
- store_data: "recalls"
- http:
uri: "/api/vehicles"
- process_vehicle_recalls: "recalls"
Custom Step
from typing import Dict, List, Union
@register_processor(arg_required=True)
def process_vehicle_recalls(
result: List[Dict[str, Union[str, int]]],
metadata: Dict[str, List[Dict[str, Union[str, int]]]],
step_args: str,
) -> List[Dict[str, Union[str, int]]]:
recall_list: List[Dict[str, Union[str, int]]] = metadata[step_args]
vehicle_info: Dict[int, Dict[str, Union[str, int]]] = {
vehicle.pop("id"): vehicle for vehicle in result
}
for recall in recall_list:
vehicle_id: int = recall.pop("vehicle_id")
vehicle_data: Dict[str, Union[str, int]] = vehicle_info.get(vehicle_id, {})
recall["vehicle_model"] = vehicle_data.get("model", "UNKNOWN")
return recall_list
Result
[
{
"recall_id": "RCL-2026-0001",
"vehicle_id": 1,
"severity": "CRITICAL",
"description": "A-pillar malfunction may reduce structural integrity in a collision.",
"issued_date": "2026-01-14",
"status": "OPEN",
"vehicle_model": "Explorer",
},
{
"recall_id": "RCL-2025-0187",
"vehicle_id": 2,
"severity": "HIGH",
"description": "Brake line corrosion may cause brake fluid leak and reduced braking performance.",
"issued_date": "2025-11-02",
"status": "OPEN",
"vehicle_model": "F-150",
},
{
"recall_id": "RCL-2024-0440",
"vehicle_id": 3,
"severity": "MEDIUM",
"description": "Airbag sensor calibration may incorrectly disable passenger airbag.",
"issued_date": "2024-06-18",
"status": "RESOLVED",
"vehicle_model": "Escape",
},
]
Dependent Requests
Sometimes an API returns a list of resource URLs that need to be individually queried to get complete information. This example demonstrates how to make an initial request to get a list of URLs, then automatically make follow-up requests to each URL and aggregate the results.
Scenario 1: Basic Dependent Requests
We have an API endpoint that returns a list of towns with URLs to their detailed information:
/api/towns- Returns a list of towns with links to individual town details
We want to fetch the list of towns, extract their URLs, make individual requests to each URL, and combine all the detailed information into a single response.
Towns Payload: /api/towns
[
{
"name": "Reston",
"url": "https://api.example.com/towns/reston",
"population": 65000
},
{
"name": "Washington D.C.",
"url": "https://api.example.com/towns/washington-dc",
"population": 700000
},
{
"name": "Durham",
"url": "https://api.example.com/towns/durham",
"population": 310000
}
]
Individual Town Detail Payload: (e.g., https://api.example.com/towns/durham)
{
"name": "Durham",
"population": 310000,
"mayor": "Leonardo Williams",
"founded": 1869,
"area_sq_miles": 115.4,
}
The snippet argument retrieves the list of towns, extracts just the URLs using JMESPath, and then uses a custom requestor to make individual requests to each URL:
Snippet Argument
low_code:
version: 2
steps:
- http:
uri: "/api/towns"
- jmespath:
value: "[].url"
- dependent_request
The JMESPath query "[].url" extracts only the URL field from each town, transforming
the data into a simple list:
[
"https://api.example.com/towns/reston",
"https://api.example.com/towns/washington-dc",
"https://api.example.com/towns/durham"
]
Custom Step
from typing import Any, Callable, Dict, List, Union
@register_requestor
def dependent_request(
result: List[str],
result_container: ResultContainer,
debug: Callable,
) -> dict[str, dict[str, Any]]:
response: dict[str, dict[str, Any]] = {}
for url in result:
http_resp = sf_perform_request(
result_container,
debug,
step_args={"url": url, "convert": True},
).converted
response[url] = http_resp
return response
Result
The final output is a dictionary where each URL serves as a key, containing the detailed response from that endpoint:
{
"https://api.example.com/towns/reston": {
"name": "Reston",
"population": 65000,
"mayor": None,
"founded": 1964,
"area_sq_miles": 15,
},
"https://api.example.com/towns/washington-dc": {
"name": "Washington D.C.",
"population": 700000,
"mayor": "Muriel Bowser",
"founded": 1790,
"area_sq_miles": 68,
},
"https://api.example.com/towns/durham": {
"name": "Durham",
"population": 310000,
"mayor": "Leonardo Williams",
"founded": 1869,
"area_sq_miles": 142.7,
}
}
Scenario 2: Dependent Requests with Context
Building on the previous example, this scenario handles more complex request configurations. Instead of simple URL strings, the API may return either a string URL or a dictionary containing a URL and optional metadata (like an API key). Additionally, we want to preserve the request context alongside the response data for better traceability.
In this scenario:
URLs can be either strings or dictionaries with a
urlfieldDictionaries may include an optional
keyfield for custom identificationThe response includes both the request context and the API result
Towns Payload: /api/towns
[
{
"name": "Reston",
"url": "https://api.example.com/towns/reston",
"population": 65000
},
{
"name": "Washington D.C.",
"url": {"url": "https://api.example.com/towns/washington-dc"},
"population": 700000
},
{
"name": "Durham",
"url": {"url": "https://api.example.com/towns/durham", "key": "durham_key"},
"population": 310000
}
]
The individual town detail payloads remain the same as Scenario 1. The snippet argument follows the same pattern: retrieve the list of towns, extracts the URLs, and use a custom requestor to make individual requests.
Snippet Argument
low_code:
version: 2
steps:
- http:
uri: "/api/towns"
- jmespath:
value: "[].url"
- dependent_request_with_context
The JMESPath query "[].url" extracts the URL field from each town. Note that the URLs
are now a mix of strings and dictionaries:
[
"https://api.example.com/towns/reston",
{"url": "https://api.example.com/towns/washington-dc"},
{"url": "https://api.example.com/towns/durham", "key": "durham_key"}
]
Custom Step
The custom step normalizes the input (converting strings to dictionaries), extracts or generates keys for identification, and preserves the request context alongside the response:
from typing import Any, Callable, Dict, List, Union
@register_requestor
def dependent_request_with_context(
result: List[Union[Dict[str, str], str]],
result_container: ResultContainer,
debug: Callable,
) -> dict[str, dict[str, Any]]:
request_info_list: list[dict[str, str] | str] = (
result if isinstance(result, list) else [result]
)
response: dict[str, dict[str, Any]] = {}
for request_info in request_info_list:
if isinstance(request_info, str):
request_info = {"url": request_info}
url = request_info["url"]
key = request_info.get("key", url)
request_info["key"] = key
response[key] = {"context": request_info}
http_resp = sf_perform_request(
result_container,
debug,
step_args={"url": url, "convert": True},
).converted
response[key]["result"] = http_resp
return response
Result
The final output is a dictionary containing the detailed response from that endpoint as well as the context used to make the request. If a key was provided, it is used as the key in the response:
{
"https://api.example.com/towns/reston": {
context: {
"url": "https://api.example.com/towns/reston",
"key": "https://api.example.com/towns/reston"
},
result: {
"name": "Reston",
"population": 65000,
"mayor": None,
"founded": 1964,
"area_sq_miles": 15,
}
},
"https://api.example.com/towns/washington-dc": {
context: {
"url": "https://api.example.com/towns/washington-dc",
"key": "https://api.example.com/towns/washington-dc"
},
result: {
"name": "Washington D.C.",
"population": 700000,
"mayor": "Muriel Bowser",
"founded": 1790,
"area_sq_miles": 68,
}
},
"durham_key": {
context: {
"url": "https://api.example.com/towns/durham",
"key": "durham_key"
},
result: {
"name": "Durham",
"population": 310000,
"mayor": "Leonardo Williams",
"founded": 1869,
"area_sq_miles": 142.7,
}
}
}