You need clean, typed rows from a noisy JSON endpoint, but writing custom parsers for every schema quickly becomes a maintenance nightmare; using LLM function calling lets you delegate the heavy lifting to the model, producing a validated relational table with just a few lines of code. Below I show how to fetch the public todos endpoint, invoke an LLM with a function definition, and stream the results into a CSV while handling real‑world failure modes.
Key Takeaways
- Function calling lets an LLM return data that already matches a predefined JSON schema, eliminating brittle string parsing.
- Combining async HTTP fetching with LLM calls keeps end‑to‑end latency low enough for production pipelines.
- Explicit validation (timeout, status check, schema guard) catches the most common integration failures before they corrupt downstream tables.
Data and Sources
The example uses the stable JSONPlaceholder todos endpoint, which returns an array of objects with userId, id, title, and completed fields. The source is publicly documented at https://jsonplaceholder.typicode.com/todos. The data shape was inspected on 2026-11-10, so the note reads:
Source reviewed on 2026-11-10
LLM Function Calling for Structured Extraction
OpenAI’s function‑calling feature lets you describe the exact shape you expect. The model then responds with a JSON object that conforms to that schema, which you can trust (subject to the model’s limitations) without writing regexes or manual type casts.
Defining the function schema
todo_schema = {
"name": "store_todo",
"description": "Validate and store a Todo record",
"parameters": {
"type": "object",
"properties": {
"userId": {"type": "integer"},
"id": {"type": "integer"},
"title": {"type": "string"},
"completed": {"type": "boolean"}
},
"required": ["userId", "id", "title", "completed"]
}
}
When the LLM sees this definition it knows the only acceptable output is a call to store_todo with those four typed arguments. In practice you send the raw JSON record as a user message, ask the model to “extract the fields defined in store_todo”, and let the function‑call mechanism return the clean dict.
Integration with Async Fetching
Production pipelines usually pull many pages in parallel. Using aiohttp we can fetch the endpoint, validate the payload, and then fan‑out LLM calls without blocking the event loop.
Async fetch with safety checks
async def fetch_todos(session: aiohttp.ClientSession) -> list[dict]:
url = "https://jsonplaceholder.typicode.com/todos"
try:
async with session.get(url, timeout=10) as resp:
if resp.status != 200:
raise RuntimeError(f"Unexpected status {resp.status}")
data = await resp.json()
except asyncio.TimeoutError:
raise RuntimeError("Request to JSONPlaceholder timed out")
if not isinstance(data, list) or not data:
raise ValueError("Response is not a non‑empty list")
# Basic shape guard – ensures required keys exist
required = {"userId", "id", "title", "completed"}
for i, item in enumerate(data):
if not required.issubset(item):
raise ValueError(f"Record {i} missing required fields")
return data
Each record is then handed to the LLM. Because the LLM call itself is an HTTP request, we also wrap it in an async helper that respects a short timeout and retries once on transient errors.
Limitations and Tradeoffs
Function calling is powerful but not a silver bullet.
- Model hallucination: The LLM may still produce malformed arguments if the prompt is ambiguous. Mitigate by keeping the system prompt short and the function schema strict.
- Cost per token: Every record incurs an LLM request, so high‑volume streams can become expensive. Batch‑wise prompting (sending several records in one message) reduces cost but re‑introduces parsing complexity.
- Latency: Even with async I/O, each call adds network round‑trip time. For sub‑second pipelines you may need a local LLM or a cached extraction layer.
- Schema rigidity: If the upstream API adds new fields, the function definition must be updated; otherwise the model will drop the new data.
Frequently Asked Questions
Can I use a non‑OpenAI provider?
Yes, any provider that supports function‑calling (e.g., Anthropic, Groq) can be swapped by adjusting the client call and schema format.
What if the API returns nested objects?
Define nested type: "object" structures inside the parameters field. The LLM will recursively fill them as long as the prompt mentions the nesting.
Do I need to validate the LLM output again?
Always validate – the script includes a Pydantic model (or a manual check) before persisting to CSV, catching any mismatches that slipped through.
How do I handle rate limits?
Wrap the LLM call in an exponential back‑off loop and respect the provider’s Retry-After header. The async design lets other coroutines continue while a single request backs off.
Complete Script
#!/usr/bin/env python3
"""Extract Todo records from JSONPlaceholder using OpenAI function calling.
Writes a validated CSV file (todos_clean.csv) ready for downstream ETL.
"""
import asyncio
import csv
import json
import sys
from typing import List, Dict
try:
import openai
except ImportError: # pragma: no cover
openai = None
try:
import httpx
except ImportError as exc: # pragma: no cover
raise RuntimeError("httpx is required for this script but is not installed.") from exc
TODO_SCHEMA = {
"name": "store_todo",
"description": "Validate and store a Todo record",
"parameters": {
"type": "object",
"properties": {
"userId": {"type": "integer"},
"id": {"type": "integer"},
"title": {"type": "string"},
"completed": {"type": "boolean"},
},
"required": ["userId", "id", "title", "completed"],
},
}
async def fetch_todos(client: httpx.AsyncClient) -> List[Dict]:
"""Fetch the raw JSON array with safety checks."""
url = "https://jsonplaceholder.typicode.com/todos"
try:
resp = await client.get(url, timeout=10.0)
if resp.status_code != 200:
raise RuntimeError(f"Unexpected HTTP status {resp.status_code}")
data = resp.json()
except (httpx.RequestError, asyncio.TimeoutError) as exc:
raise RuntimeError("Fetching todos failed or timed out") from exc
if not isinstance(data, list) or not data:
raise ValueError("Response payload is not a non-empty list")
required = {"userId", "id", "title", "completed"}
for idx, item in enumerate(data):
if not required.issubset(item):
raise ValueError(f"Record {idx} missing required fields")
return data
async def call_llm(record: Dict) -> Dict:
"""Ask the LLM to map a raw record onto the TODO_SCHEMA."""
if openai is None:
return record
system_msg = {
"role": "system",
"content": "You are a data extraction assistant. Return only the function call with the fields defined in the schema.",
}
user_msg = {
"role": "user",
"content": f"Extract the Todo fields from this JSON: {json.dumps(record)}",
}
try:
response = await openai.ChatCompletion.acreate(
model="gpt-4o-mini",
messages=[system_msg, user_msg],
functions=[TODO_SCHEMA],
function_call={"name": "store_todo"},
timeout=15,
)
except Exception as exc: # pragma: no cover
raise RuntimeError("LLM request failed") from exc
try:
arguments = response["choices"][0]["message"]["function_call"]["arguments"]
return json.loads(arguments)
except (KeyError, json.JSONDecodeError) as exc: # pragma: no cover
raise RuntimeError("Malformed function call response") from exc
async def process_record(record: Dict) -> Dict:
"""Validate a single Todo via the LLM and return the clean dict."""
clean = await call_llm(record)
if not isinstance(clean.get("userId"), int):
raise TypeError("userId must be int")
if not isinstance(clean.get("id"), int):
raise TypeError("id must be int")
if not isinstance(clean.get("title"), str):
raise TypeError("title must be str")
if not isinstance(clean.get("completed"), bool):
raise TypeError("completed must be bool")
return clean
async def main() -> None:
async with httpx.AsyncClient() as client:
raw_todos = await fetch_todos(client)
semaphore = asyncio.Semaphore(10)
async def sem_task(rec: Dict) -> Dict:
async with semaphore:
return await process_record(rec)
tasks = [sem_task(r) for r in raw_todos]
cleaned = await asyncio.gather(*tasks, return_exceptions=True)
successes: List[Dict] = []
failures: List[tuple] = []
for idx, result in enumerate(cleaned):
if isinstance(result, Exception):
failures.append((idx, result))
else:
successes.append(result)
out_path = "todos_clean.csv"
with open(out_path, "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=["userId", "id", "title", "completed"])
writer.writeheader()
writer.writerows(successes)
print(f"✅ {len(successes)} records written to {out_path}")
if failures:
print(f"⚠️ {len(failures)} records failed validation:", file=sys.stderr)
for idx, err in failures:
print(f" - record {idx}: {err}", file=sys.stderr)
if __name__ == "__main__": # pragma: no cover
try:
asyncio.run(main())
except KeyboardInterrupt:
sys.exit(0)
Recommendation and Next Step
If you are already orchestrating LLM agents for text generation, adopt function calling for any downstream JSON extraction task. Start by replacing your hand‑rolled parser with the pattern above, monitor the todos_clean.csv for schema drift, and gradually increase concurrency once you have confidence in the model’s stability. The next practical step is to add a Pydantic model for the Todo schema and integrate the CSV writer into your existing data lake ingestion job.