> ## Documentation Index
> Fetch the complete documentation index at: https://docs.perplexity.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Enrich a CSV of Companies Overnight with the Batch API

> Write one web-grounded Agent API request per CSV row to a JSONL file, run them asynchronously with the Batch API, and join the output back to your rows by custom_id.

Enriching hundreds of company rows should not keep your application waiting: use the Batch API to run web-grounded Agent API requests asynchronously, then join the results back to your CSV. This recipe fills five columns: headquarters city, founding year, a one-line description, a recent news headline, and its URL. You submit the work once and keep completed rows even when some requests fail.

The [Batch API](/docs/agent-api/batch.md) is built for this. You write the same Agent API requests to a JSONL file, upload it, create a batch, and come back later. Requests inside a batch do not count toward the Agent API request rate limit (token rate limits still apply, and the batch retries a request that hits one). They run on the `flex` [service tier](/docs/agent-api/models.md#service-tiers) where the model supports it, including `openai/gpt-5.6-sol`. Flex halves model token prices only; tool calls such as `web_search` bill at their normal price. When the batch ends you download an output file, one JSON line per request, and join it back to your rows by the `custom_id` you chose.

This recipe does that for a 12-row sample CSV. `enrichment.py` builds the request file and validates the results; `run_batch.py` handles upload, polling, and download. Two test files check the parser and batch lifecycle without an API key. The model returns JSON text for each company; your code validates the field types, joins it to the row, and decides what to retry.

## What you need

* Python 3.10 or newer on macOS or Linux. The shell commands use Unix syntax.
* A Perplexity API key with API credits. Organizations with Zero Data Retention cannot use batch, because batch keeps inputs and outputs for 30 days; for them the endpoints return `403`.
* `httpx`, a plain HTTP client. This recipe calls the four endpoints directly: `POST /v1/files`, `POST /v1/batches`, `GET /v1/batches/{id}`, and `GET /v1/files/{id}/content`. You do not need the Perplexity Python library.

Every line in a batch file must use the same `model`. This recipe uses `openai/gpt-5.6-sol`; see the [Batch API guide](/docs/agent-api/batch.md#supported-requests) for other supported models and request restrictions.

## Set up

Everything lives in one directory. You create the first six files from this page; the script writes the rest.

```text theme={null}
batch-api-company-enrichment/
├── requirements.txt          # two pinned dependencies
├── companies.csv             # 12 sample rows: name, website
├── enrichment.py             # build the JSONL, parse results, join, retry file
├── run_batch.py              # upload, create, poll, download, summarize
├── test_enrichment.py        # offline tests, no API key
├── test_run_batch.py         # offline lifecycle and resume tests
├── batch_input.jsonl         # written by run_batch.py: one request per row
├── batch_state.json          # written by run_batch.py: file id, batch id
├── batch_output.jsonl        # downloaded: requests that ran to the end
├── batch_errors.jsonl        # downloaded: requests that did not
├── batch_retry.jsonl         # written by run_batch.py: requests to send again
├── companies_enriched.csv    # the join: every input row plus the new columns
└── .venv/                    # virtual environment from the install step
```

> **requirements.txt**
>
> ```text requirements.txt theme={null}
> httpx==0.28.1
> pytest==8.4.2
> ```

Create the directory, save `requirements.txt` in it, then create a virtual environment, install the packages, and export your key.

> **Install and set the key**
>
> ```bash theme={null}
> mkdir batch-api-company-enrichment && cd batch-api-company-enrichment
> # save requirements.txt here, then continue
> python3 -m venv .venv
> source .venv/bin/activate
> python -m pip install -r requirements.txt
> export PERPLEXITY_API_KEY="your-api-key-here"
> ```

## The companies

Twelve well-known public companies with a name and a website. The website is in the prompt so the search lands on the right Apple. Save the file as `companies.csv`. Your real file can include additional columns.

The script replaces any existing enrichment or status columns and preserves your other columns. Schema validation checks the shape and types, not whether the claims are true or the news URL is trustworthy. Review returned sources before using the rows for customer-facing claims, and do not treat a web search as an exhaustive check for the latest news.

> **companies.csv**
>
> ```text companies.csv theme={null}
> name,website
> Apple,https://www.apple.com
> Microsoft,https://www.microsoft.com
> Nvidia,https://www.nvidia.com
> Shopify,https://www.shopify.com
> Toyota,https://global.toyota
> Nestlé,https://www.nestle.com
> Siemens,https://www.siemens.com
> Spotify,https://www.spotify.com
> Atlassian,https://www.atlassian.com
> Salesforce,https://www.salesforce.com
> Netflix,https://www.netflix.com
> Airbnb,https://www.airbnb.com
> ```

## Build the request file

`enrichment.py` holds everything that does not touch the network. Each excerpt below is an exact slice of the file, which is at the end of the section.

### Constants, schema, and instructions

`MODEL` is the one model every line uses. `RESPONSE_SCHEMA` is the JSON shape you want back, sent as `response_format` so the Agent API returns an object with exactly these five keys ([Output Control](/docs/agent-api/output-control.md#structured-outputs)). `INSTRUCTIONS` tells the model to confirm every field with web search and to return only the object. The schema is the contract; the instructions improve how often the model meets it.

> **Constants, schema, and instructions**
>
> ```python theme={null}
> """Build batch requests, validate results, join CSV rows, and write retries without network calls."""
>
> from __future__ import annotations
>
> import csv
> import json
> import re
> from dataclasses import dataclass, field
> from typing import Any
>
> MODEL = "openai/gpt-5.6-sol"  # one model per file; every line must match
> BATCH_URL = "/v1/responses"  # the only url the Batch API accepts
> CUSTOM_ID_MAX_BYTES = 64
>
> ENRICHMENT_FIELDS = (
>     "headquarters_city",
>     "founded_year",
>     "description",
>     "latest_news_headline",
>     "latest_news_url",
> )
> RESPONSE_SCHEMA: dict[str, Any] = {
>     "type": "object",
>     "properties": {
>         "headquarters_city": {"type": "string"},
>         "founded_year": {"type": "integer"},
>         "description": {"type": "string"},
>         "latest_news_headline": {"type": "string"},
>         "latest_news_url": {"type": "string"},
>     },
>     "required": list(ENRICHMENT_FIELDS),
>     "additionalProperties": False,
> }
> INSTRUCTIONS = (
>     "You are a research assistant filling in a company database. Use web search "
>     "to confirm every field against a current source. For the news fields, use the "
>     "most recent news article about the company and return that article's URL. "
>     "Return only the JSON object; no markdown, no commentary."
> )
> ```

### Custom ids and the CSV

`custom_id` is how you get results back to rows, because the output file is not in input order. It must be unique within the file and 1 to 64 bytes. `make_custom_id` builds `row-003-nvidia` from the row number and a slug of the name, and truncates the slug on a UTF-8 boundary so a long or accented name cannot push it over 64 bytes. `read_companies` reads the CSV and attaches the id to every row.

> **make\_custom\_id() and read\_companies()**
>
> ```python theme={null}
> def make_custom_id(index: int, name: str) -> str:
>     """Build a unique, readable id of at most 64 bytes: `row-003-nvidia`."""
>     slug = re.sub(r"[^a-z0-9]+", "-", name.lower()).strip("-") or "row"
>     prefix = f"row-{index:03d}-"
>     budget = CUSTOM_ID_MAX_BYTES - len(prefix)
>     return prefix + slug.encode("utf-8")[:budget].decode("utf-8", "ignore")
>
>
> def read_companies(path: str) -> list[dict[str, str]]:
>     """Read the CSV and attach a custom_id to every row."""
>     with open(path, newline="", encoding="utf-8") as handle:
>         rows = [dict(row) for row in csv.DictReader(handle)]
>     for index, row in enumerate(rows, start=1):
>         row["custom_id"] = make_custom_id(index, row["name"])
>     return rows
> ```

### One request per row

`build_request_body` is a normal Agent API request body: the model, the instructions, a prompt with the company name and website, the built-in [`web_search`](/docs/agent-api/tools/web-search) tool, and the `response_format`. The same function serves a batch line and a single synchronous `POST /v1/agent` call, which is how you spot-check a prompt change before you send 800 of them.

`build_batch_lines` wraps each body in the batch envelope: `custom_id`, `method: POST`, `url: /v1/responses`. `write_jsonl` writes one compact object per line with LF endings and no blank lines, which is what the upload validates.

> **build\_request\_body(), build\_batch\_lines(), and the JSONL helpers**
>
> ```python theme={null}
> def build_request_body(row: dict[str, str]) -> dict[str, Any]:
>     """One Agent API request body. Shared by the batch file and a live spot check."""
>     prompt = (
>         f"Company: {row['name']}\nWebsite: {row['website']}\n\n"
>         "Find the headquarters city, the founding year, a one-sentence description "
>         "of what the company does, and the most recent news headline about the "
>         "company with the URL of that article."
>     )
>     return {
>         "model": MODEL,
>         "instructions": INSTRUCTIONS,
>         "input": prompt,
>         "tools": [{"type": "web_search"}],
>         "response_format": {
>             "type": "json_schema",
>             "json_schema": {"name": "company_enrichment", "schema": RESPONSE_SCHEMA},
>         },
>     }
>
>
> def build_batch_lines(rows: list[dict[str, str]]) -> list[dict[str, Any]]:
>     """One batch request per row: custom_id, method, url, body."""
>     return [
>         {
>             "custom_id": row["custom_id"],
>             "method": "POST",
>             "url": BATCH_URL,
>             "body": build_request_body(row),
>         }
>         for row in rows
>     ]
>
>
> def write_jsonl(lines: list[dict[str, Any]], path: str) -> None:
>     """One compact JSON object per line, LF endings, no blank lines."""
>     with open(path, "w", encoding="utf-8", newline="\n") as handle:
>         handle.writelines(json.dumps(line, ensure_ascii=False) + "\n" for line in lines)
>
>
> def read_jsonl(path: str) -> list[dict[str, Any]]:
>     with open(path, encoding="utf-8") as handle:
>         return [json.loads(line) for line in handle if line.strip()]
> ```

## Parse the results

### What the recipe keeps per row

`Result` is one row's outcome: a status, the five parsed fields, an error string, the Response id, and the API-reported cost with its included tool subtotal. `ok` is true only for `completed`; everything else goes to the retry file. `message_text` concatenates the `output_text` parts of every `message` item in a Response. `parse_enrichment` loads the JSON, strips a markdown fence if the model added one anyway, and returns `None` unless all five keys are present.

> **Result, message\_text(), and parse\_enrichment()**
>
> ```python theme={null}
> @dataclass
> class Result:
>     """What the recipe keeps for one row after the batch finishes."""
>
>     custom_id: str
>     status: str  # completed | incomplete | failed | error | parse_error | missing
>     fields: dict[str, Any] = field(default_factory=dict)
>     error: str = ""
>     response_id: str = ""
>     cost_usd: float = 0.0
>     tool_calls_cost_usd: float = 0.0
>
>     @property
>     def ok(self) -> bool:
>         return self.status == "completed"
>
>
> def message_text(body: dict[str, Any]) -> str:
>     """Concatenate the output_text parts of every message in a Response."""
>     parts: list[str] = []
>     for item in body.get("output") or []:
>         if item.get("type") != "message":
>             continue
>         for part in item.get("content") or []:
>             if part.get("type") == "output_text":
>                 parts.append(part.get("text", ""))
>     return "".join(parts)
>
>
> def parse_enrichment(text: str) -> dict[str, Any] | None:
>     """Parse the model's JSON. Return None when it is not the object you asked for."""
>     cleaned = text.strip()
>     if cleaned.startswith("`"):  # strip a markdown fence if the model added one anyway
>         cleaned = re.sub(r"^`+(?:json)?\s*|\s*`+$", "", cleaned)
>     try:
>         data = json.loads(cleaned)
>     except json.JSONDecodeError:
>         return None
>     if not isinstance(data, dict) or set(data) != set(ENRICHMENT_FIELDS):
>         return None
>     if type(data["founded_year"]) is not int:
>         return None
>     if any(not isinstance(data[k], str) for k in ENRICHMENT_FIELDS if k != "founded_year"):
>         return None
>     return {k: data[k] for k in ENRICHMENT_FIELDS}
> ```

### Output lines and error lines

An output line carries the full Response under `response.body`. `parse_output_line` reads `response.body.status` first: `incomplete` (for example `max_output_tokens` reached) and `failed` are not successes even though the line is in the output file. Only a `completed` response is parsed, and a response whose text is not the object you asked for becomes `parse_error`.

The error file has two line shapes, and `parse_error_line` handles both. A request the API rejected has `error: null` and a `response` with the HTTP status and error body. A request that stopped without an answer has `response: null` and an `error` with a code such as `batch_expired`, `batch_cancelled`, `poll_timeout`, or `response_too_large`.

> **parse\_output\_line() and parse\_error\_line()**
>
> ```python theme={null}
> def parse_output_line(line: dict[str, Any]) -> Result:
>     """One line of the output file: a full Response under response.body."""
>     body = (line.get("response") or {}).get("body") or {}
>     status = body.get("status", "failed")
>     cost = (body.get("usage") or {}).get("cost") or {}
>     result = Result(
>         custom_id=line["custom_id"],
>         status=status,
>         response_id=body.get("id", ""),
>         cost_usd=float(cost.get("total_cost") or 0.0),
>         tool_calls_cost_usd=float(cost.get("tool_calls_cost") or 0.0),
>     )
>     if status != "completed":
>         error = body.get("error") or {}
>         result.error = error.get("message", f"response status {status}")
>         return result
>     fields = parse_enrichment(message_text(body))
>     if fields is None:
>         result.status = "parse_error"
>         result.error = "response text was not the expected JSON object"
>         return result
>     result.fields = fields
>     return result
>
>
> def parse_error_line(line: dict[str, Any]) -> Result:
>     """One line of the error file. Two shapes: an API rejection, or a stopped run."""
>     response = line.get("response")
>     if response:  # the API rejected the request: status_code plus an error body
>         err = (response.get("body") or {}).get("error") or {}
>         message = err.get("message", "request rejected")
>         return Result(
>             custom_id=line["custom_id"],
>             status="error",
>             error=f"HTTP {response.get('status_code')}: {message}",
>         )
>     err = line.get("error") or {}  # the run stopped: batch_expired, poll_timeout, ...
>     return Result(
>         custom_id=line["custom_id"],
>         status="error",
>         error=f"{err.get('code', 'unknown')}: {err.get('message', '')}".strip(": "),
>     )
> ```

### Join, retry file, and summary

`collect_results` indexes every output and error line by `custom_id`, then walks your rows and marks any row with no line at all as `missing`. That happens when a batch ends `failed` and `output_file_id` is `null`, and it means no row can silently disappear from the join. `write_enriched_csv` writes every input row with its original columns, the five new ones, and `status`, `error`, and `response_id`. `retry_lines` returns the original request lines for every row that is not `completed`, so the retry file is ready to upload as the next batch. `summarize` counts rows by status and sums the per-request cost from the output file.

> **collect\_results(), write\_enriched\_csv(), retry\_lines(), and summarize()**
>
> ```python theme={null}
> def collect_results(
>     rows: list[dict[str, str]],
>     output_lines: list[dict[str, Any]],
>     error_lines: list[dict[str, Any]],
> ) -> dict[str, Result]:
>     """Index every result by custom_id. Rows with no line at all become `missing`."""
>     results: dict[str, Result] = {}
>     for line in output_lines:
>         results[line["custom_id"]] = parse_output_line(line)
>     for line in error_lines:
>         results[line["custom_id"]] = parse_error_line(line)
>     for row in rows:
>         results.setdefault(
>             row["custom_id"],
>             Result(row["custom_id"], "missing", error="no output or error line"),
>         )
>     return results
>
>
> def write_enriched_csv(rows: list[dict[str, str]], results: dict[str, Result], path: str) -> None:
>     """Join by custom_id and keep every input row, enriched or not."""
>     base_columns = [c for c in rows[0] if c != "custom_id"] if rows else []
>     columns = list(dict.fromkeys(base_columns + list(ENRICHMENT_FIELDS) + ["status", "error", "response_id"]))
>     with open(path, "w", newline="", encoding="utf-8") as handle:
>         writer = csv.DictWriter(handle, fieldnames=columns)
>         writer.writeheader()
>         for row in rows:
>             result = results[row["custom_id"]]
>             out = {c: row.get(c, "") for c in base_columns}
>             out.update({k: result.fields.get(k, "") for k in ENRICHMENT_FIELDS})
>             out["status"], out["error"] = result.status, result.error
>             out["response_id"] = result.response_id
>             writer.writerow(out)
>
>
> def retry_lines(batch_lines: list[dict], results: dict[str, Result]) -> list[dict]:
>     """The original request lines for every row that did not complete."""
>     failed = {cid for cid, result in results.items() if not result.ok}
>     return [line for line in batch_lines if line["custom_id"] in failed]
>
>
> def summarize(results: dict[str, Result]) -> dict[str, Any]:
>     """Counts by status plus the summed per-request cost from the output file."""
>     counts: dict[str, int] = {}
>     for result in results.values():
>         counts[result.status] = counts.get(result.status, 0) + 1
>     return {
>         "total": len(results),
>         "completed": counts.get("completed", 0),
>         "needs_retry": sum(1 for r in results.values() if not r.ok),
>         "by_status": dict(sorted(counts.items())),
>         "cost_usd": round(sum(r.cost_usd for r in results.values()), 4),
>         "tool_calls_cost_usd": round(sum(r.tool_calls_cost_usd for r in results.values()), 4),
>     }
> ```

### Complete enrichment.py

Save this as `enrichment.py`. It is the six excerpts above in order, nothing else.

> **Full code - enrichment.py**
>
> ```python enrichment.py theme={null}
> """Build batch requests, validate results, join CSV rows, and write retries without network calls."""
>
> from __future__ import annotations
>
> import csv
> import json
> import re
> from dataclasses import dataclass, field
> from typing import Any
>
> MODEL = "openai/gpt-5.6-sol"  # one model per file; every line must match
> BATCH_URL = "/v1/responses"  # the only url the Batch API accepts
> CUSTOM_ID_MAX_BYTES = 64
>
> ENRICHMENT_FIELDS = (
>     "headquarters_city",
>     "founded_year",
>     "description",
>     "latest_news_headline",
>     "latest_news_url",
> )
> RESPONSE_SCHEMA: dict[str, Any] = {
>     "type": "object",
>     "properties": {
>         "headquarters_city": {"type": "string"},
>         "founded_year": {"type": "integer"},
>         "description": {"type": "string"},
>         "latest_news_headline": {"type": "string"},
>         "latest_news_url": {"type": "string"},
>     },
>     "required": list(ENRICHMENT_FIELDS),
>     "additionalProperties": False,
> }
> INSTRUCTIONS = (
>     "You are a research assistant filling in a company database. Use web search "
>     "to confirm every field against a current source. For the news fields, use the "
>     "most recent news article about the company and return that article's URL. "
>     "Return only the JSON object; no markdown, no commentary."
> )
>
>
> def make_custom_id(index: int, name: str) -> str:
>     """Build a unique, readable id of at most 64 bytes: `row-003-nvidia`."""
>     slug = re.sub(r"[^a-z0-9]+", "-", name.lower()).strip("-") or "row"
>     prefix = f"row-{index:03d}-"
>     budget = CUSTOM_ID_MAX_BYTES - len(prefix)
>     return prefix + slug.encode("utf-8")[:budget].decode("utf-8", "ignore")
>
>
> def read_companies(path: str) -> list[dict[str, str]]:
>     """Read the CSV and attach a custom_id to every row."""
>     with open(path, newline="", encoding="utf-8") as handle:
>         rows = [dict(row) for row in csv.DictReader(handle)]
>     for index, row in enumerate(rows, start=1):
>         row["custom_id"] = make_custom_id(index, row["name"])
>     return rows
>
>
> def build_request_body(row: dict[str, str]) -> dict[str, Any]:
>     """One Agent API request body. Shared by the batch file and a live spot check."""
>     prompt = (
>         f"Company: {row['name']}\nWebsite: {row['website']}\n\n"
>         "Find the headquarters city, the founding year, a one-sentence description "
>         "of what the company does, and the most recent news headline about the "
>         "company with the URL of that article."
>     )
>     return {
>         "model": MODEL,
>         "instructions": INSTRUCTIONS,
>         "input": prompt,
>         "tools": [{"type": "web_search"}],
>         "response_format": {
>             "type": "json_schema",
>             "json_schema": {"name": "company_enrichment", "schema": RESPONSE_SCHEMA},
>         },
>     }
>
>
> def build_batch_lines(rows: list[dict[str, str]]) -> list[dict[str, Any]]:
>     """One batch request per row: custom_id, method, url, body."""
>     return [
>         {
>             "custom_id": row["custom_id"],
>             "method": "POST",
>             "url": BATCH_URL,
>             "body": build_request_body(row),
>         }
>         for row in rows
>     ]
>
>
> def write_jsonl(lines: list[dict[str, Any]], path: str) -> None:
>     """One compact JSON object per line, LF endings, no blank lines."""
>     with open(path, "w", encoding="utf-8", newline="\n") as handle:
>         handle.writelines(json.dumps(line, ensure_ascii=False) + "\n" for line in lines)
>
>
> def read_jsonl(path: str) -> list[dict[str, Any]]:
>     with open(path, encoding="utf-8") as handle:
>         return [json.loads(line) for line in handle if line.strip()]
>
>
> @dataclass
> class Result:
>     """What the recipe keeps for one row after the batch finishes."""
>
>     custom_id: str
>     status: str  # completed | incomplete | failed | error | parse_error | missing
>     fields: dict[str, Any] = field(default_factory=dict)
>     error: str = ""
>     response_id: str = ""
>     cost_usd: float = 0.0
>     tool_calls_cost_usd: float = 0.0
>
>     @property
>     def ok(self) -> bool:
>         return self.status == "completed"
>
>
> def message_text(body: dict[str, Any]) -> str:
>     """Concatenate the output_text parts of every message in a Response."""
>     parts: list[str] = []
>     for item in body.get("output") or []:
>         if item.get("type") != "message":
>             continue
>         for part in item.get("content") or []:
>             if part.get("type") == "output_text":
>                 parts.append(part.get("text", ""))
>     return "".join(parts)
>
>
> def parse_enrichment(text: str) -> dict[str, Any] | None:
>     """Parse the model's JSON. Return None when it is not the object you asked for."""
>     cleaned = text.strip()
>     if cleaned.startswith("`"):  # strip a markdown fence if the model added one anyway
>         cleaned = re.sub(r"^`+(?:json)?\s*|\s*`+$", "", cleaned)
>     try:
>         data = json.loads(cleaned)
>     except json.JSONDecodeError:
>         return None
>     if not isinstance(data, dict) or set(data) != set(ENRICHMENT_FIELDS):
>         return None
>     if type(data["founded_year"]) is not int:
>         return None
>     if any(not isinstance(data[k], str) for k in ENRICHMENT_FIELDS if k != "founded_year"):
>         return None
>     return {k: data[k] for k in ENRICHMENT_FIELDS}
>
>
> def parse_output_line(line: dict[str, Any]) -> Result:
>     """One line of the output file: a full Response under response.body."""
>     body = (line.get("response") or {}).get("body") or {}
>     status = body.get("status", "failed")
>     cost = (body.get("usage") or {}).get("cost") or {}
>     result = Result(
>         custom_id=line["custom_id"],
>         status=status,
>         response_id=body.get("id", ""),
>         cost_usd=float(cost.get("total_cost") or 0.0),
>         tool_calls_cost_usd=float(cost.get("tool_calls_cost") or 0.0),
>     )
>     if status != "completed":
>         error = body.get("error") or {}
>         result.error = error.get("message", f"response status {status}")
>         return result
>     fields = parse_enrichment(message_text(body))
>     if fields is None:
>         result.status = "parse_error"
>         result.error = "response text was not the expected JSON object"
>         return result
>     result.fields = fields
>     return result
>
>
> def parse_error_line(line: dict[str, Any]) -> Result:
>     """One line of the error file. Two shapes: an API rejection, or a stopped run."""
>     response = line.get("response")
>     if response:  # the API rejected the request: status_code plus an error body
>         err = (response.get("body") or {}).get("error") or {}
>         message = err.get("message", "request rejected")
>         return Result(
>             custom_id=line["custom_id"],
>             status="error",
>             error=f"HTTP {response.get('status_code')}: {message}",
>         )
>     err = line.get("error") or {}  # the run stopped: batch_expired, poll_timeout, ...
>     return Result(
>         custom_id=line["custom_id"],
>         status="error",
>         error=f"{err.get('code', 'unknown')}: {err.get('message', '')}".strip(": "),
>     )
>
>
> def collect_results(
>     rows: list[dict[str, str]],
>     output_lines: list[dict[str, Any]],
>     error_lines: list[dict[str, Any]],
> ) -> dict[str, Result]:
>     """Index every result by custom_id. Rows with no line at all become `missing`."""
>     results: dict[str, Result] = {}
>     for line in output_lines:
>         results[line["custom_id"]] = parse_output_line(line)
>     for line in error_lines:
>         results[line["custom_id"]] = parse_error_line(line)
>     for row in rows:
>         results.setdefault(
>             row["custom_id"],
>             Result(row["custom_id"], "missing", error="no output or error line"),
>         )
>     return results
>
>
> def write_enriched_csv(rows: list[dict[str, str]], results: dict[str, Result], path: str) -> None:
>     """Join by custom_id and keep every input row, enriched or not."""
>     base_columns = [c for c in rows[0] if c != "custom_id"] if rows else []
>     columns = list(dict.fromkeys(base_columns + list(ENRICHMENT_FIELDS) + ["status", "error", "response_id"]))
>     with open(path, "w", newline="", encoding="utf-8") as handle:
>         writer = csv.DictWriter(handle, fieldnames=columns)
>         writer.writeheader()
>         for row in rows:
>             result = results[row["custom_id"]]
>             out = {c: row.get(c, "") for c in base_columns}
>             out.update({k: result.fields.get(k, "") for k in ENRICHMENT_FIELDS})
>             out["status"], out["error"] = result.status, result.error
>             out["response_id"] = result.response_id
>             writer.writerow(out)
>
>
> def retry_lines(batch_lines: list[dict], results: dict[str, Result]) -> list[dict]:
>     """The original request lines for every row that did not complete."""
>     failed = {cid for cid, result in results.items() if not result.ok}
>     return [line for line in batch_lines if line["custom_id"] in failed]
>
>
> def summarize(results: dict[str, Result]) -> dict[str, Any]:
>     """Counts by status plus the summed per-request cost from the output file."""
>     counts: dict[str, int] = {}
>     for result in results.values():
>         counts[result.status] = counts.get(result.status, 0) + 1
>     return {
>         "total": len(results),
>         "completed": counts.get("completed", 0),
>         "needs_retry": sum(1 for r in results.values() if not r.ok),
>         "by_status": dict(sorted(counts.items())),
>         "cost_usd": round(sum(r.cost_usd for r in results.values()), 4),
>         "tool_calls_cost_usd": round(sum(r.tool_calls_cost_usd for r in results.values()), 4),
>     }
> ```

## Run the batch

`run_batch.py` is the HTTP side. It saves the file id and batch id to `batch_state.json`, and persists the create request's job name and idempotency key before sending it. Resume with `--resume` and the same input file. If a create response was lost, resume within the API's 24-hour idempotency window; after that, inspect your existing batches before creating another.

### Errors and the upload

`check` turns an HTTP error into a `BatchError` with the step name and response body. A `404` means the batch or file was not found, belongs to another organization, or has expired.

`upload_input` posts the file as `multipart/form-data` with `purpose=batch`. The API validates every line before it stores anything. If a line is bad, the upload returns `400` with the line number and reason, and you fix the file rather than discovering the problem a day later.

> **Imports, constants, and check()**
>
> ```python theme={null}
> """Enrich companies.csv overnight with the Batch API.
>
> Steps: read the CSV, write batch_input.jsonl, upload it, create a batch, poll
> until a final status, download the output and error files, join by custom_id
> into companies_enriched.csv, write a retry file, and print a summary.
>
> State (file id, batch id, idempotency key) is saved to batch_state.json after
> each step, so `--resume` picks up where a crash or Ctrl-C left off.
> """
>
> from __future__ import annotations
>
> import argparse
> import hashlib
> import json
> import os
> import sys
> import time
> from typing import Any
>
> import enrichment as en
> import httpx
>
> API = "https://api.perplexity.ai"
> FINAL_STATUSES = {"completed", "expired", "cancelled", "failed"}
> POLL_START_SECONDS = 30
> POLL_MAX_SECONDS = 300
>
>
> class BatchError(Exception):
>     """A batch or file endpoint answered with an error."""
>
>
> def check(response: httpx.Response, step: str) -> httpx.Response:
>     """Raise a readable BatchError on any non-2xx answer."""
>     if response.status_code == 404:
>         raise BatchError(
>             f"{step}: 404. The batch or file was not found, belongs to another "
>             "organization, or has expired. See the Batch API guide."
>         )
>     if response.status_code >= 400:
>         raise BatchError(f"{step}: HTTP {response.status_code} {response.text.strip()}")
>     return response
> ```

### Create the batch

`create_batch` posts the file id with `endpoint: /v1/responses` and a `metadata.job` label. The `Idempotency-Key` header makes the create safe to retry within 24 hours. The script saves both the key and metadata before sending the request, so a resume on the next day does not change the request if the original response was lost. Reusing a key with a different request returns `400`.

> **upload\_input() and create\_batch()**
>
> ```python theme={null}
> def upload_input(client: httpx.Client, path: str) -> str:
>     """POST /v1/files with purpose=batch. The API validates every line first."""
>     with open(path, "rb") as handle:
>         response = client.post(
>             f"{API}/v1/files",
>             data={"purpose": "batch"},
>             files={"file": (os.path.basename(path), handle, "application/jsonl")},
>         )
>     return check(response, "upload").json()["id"]
>
>
> def create_batch(
>     client: httpx.Client, file_id: str, idempotency_key: str, job: str
> ) -> dict[str, Any]:
>     """POST /v1/batches. A retried create with the same Idempotency-Key returns the same batch."""
>     response = client.post(
>         f"{API}/v1/batches",
>         headers={"Idempotency-Key": idempotency_key},
>         json={
>             "input_file_id": file_id,
>             "endpoint": en.BATCH_URL,
>             "completion_window": "24h",
>             "metadata": {"job": job},
>         },
>     )
>     return check(response, "create batch").json()
> ```

### Poll and download

`poll_batch` reads `GET /v1/batches/{id}` until the status is `completed`, `expired`, `cancelled`, or `failed`. It starts at 30 seconds between polls and backs off by half each time, up to five minutes, because a batch that takes hours does not need a request every 30 seconds and the batch endpoints have normal request rate limits. A `429` waits for `Retry-After`. A network error is logged and retried rather than ending the run, since the batch is still running on the server either way.

A final status means processing ended, not that every request succeeded. `download_file` fetches `output_file_id` and, when it is not `null`, `error_file_id`, writes each to disk so you keep the raw evidence, and parses the lines.

> **poll\_batch() and download\_file()**
>
> ```python theme={null}
> def poll_batch(client: httpx.Client, batch_id: str) -> dict[str, Any]:
>     """GET /v1/batches/{id} with backoff until the status is final."""
>     delay = POLL_START_SECONDS
>     while True:
>         try:
>             response = client.get(f"{API}/v1/batches/{batch_id}")
>             if response.status_code == 429:
>                 wait = float(response.headers.get("Retry-After", delay))
>                 print(f"  rate limited while polling; waiting {wait:.0f}s")
>                 time.sleep(wait)
>                 continue
>             batch = check(response, "poll").json()
>         except httpx.HTTPError as error:  # network blip; keep the overnight job alive
>             print(f"  poll failed ({error}); retrying in {delay}s")
>             time.sleep(delay)
>             continue
>         counts = batch["request_counts"]
>         print(
>             f"  {time.strftime('%H:%M:%S')} {batch['status']}: "
>             f"{counts['completed']} done, {counts['failed']} failed, {counts['total']} total"
>         )
>         if batch["status"] in FINAL_STATUSES:
>             return batch
>         time.sleep(delay)
>         delay = min(int(delay * 1.5), POLL_MAX_SECONDS)
>
>
> def download_file(client: httpx.Client, file_id: str | None, path: str) -> list[dict]:
>     """GET /v1/files/{id}/content to disk, then parse it. No id means no lines."""
>     if not file_id:
>         with open(path, "wb") as handle:
>             handle.write(b"")
>         return []
>     response = client.get(f"{API}/v1/files/{file_id}/content")
>     with open(path, "wb") as handle:
>         handle.write(check(response, f"download {file_id}").content)
>     return en.read_jsonl(path)
> ```

### The run

`main` reads the CSV, writes `batch_input.jsonl`, and hashes it. With `--requests` it uploads an existing JSONL instead, such as `batch_retry.jsonl`, and keeps only the rows in it. With `--resume` it loads `batch_state.json`, refuses to continue if the state belongs to a different input file, and skips the upload or create steps it already finished. Then it polls, downloads, joins, writes the retry file, and prints the summary: counts by status, the batch's token usage, the summed per-request cost, how long the batch took, and one line per row that needs a retry.

The batch object's `usage` covers the output file only, so it is not your full bill. In all 14 captured responses, `response.body.usage.cost.total_cost` equaled `input_cost + cache_creation_cost + output_cost + tool_calls_cost`. The script sums `total_cost` and displays `tool_calls_cost` as an included subtotal, not an extra charge. Adding that subtotal again would double-count tools in these responses.

> **load\_state(), save\_state(), and main()**
>
> ```python theme={null}
> def load_state(path: str) -> dict[str, Any]:
>     if os.path.exists(path):
>         with open(path, encoding="utf-8") as handle:
>             return json.load(handle)
>     return {}
>
>
> def save_state(path: str, state: dict[str, Any]) -> None:
>     with open(path, "w", encoding="utf-8") as handle:
>         json.dump(state, handle, indent=2)
>
>
> def main() -> int:
>     parser = argparse.ArgumentParser(description=__doc__)
>     parser.add_argument("--csv", default="companies.csv")
>     parser.add_argument("--out", default="companies_enriched.csv")
>     parser.add_argument("--requests", help="upload this JSONL (for example batch_retry.jsonl)")
>     parser.add_argument("--job", default=f"company-enrichment-{time.strftime('%Y-%m-%d')}")
>     parser.add_argument("--resume", action="store_true", help="reuse batch_state.json")
>     args = parser.parse_args()
>
>     api_key = os.environ.get("PERPLEXITY_API_KEY")
>     if not api_key:
>         print("Set PERPLEXITY_API_KEY first.", file=sys.stderr)
>         return 2
>
>     started = time.monotonic()
>     rows = en.read_companies(args.csv)
>     if args.requests:  # a retry run: upload an existing file, keep only its rows
>         input_path = args.requests
>         batch_lines = en.read_jsonl(input_path)
>         wanted = {line["custom_id"] for line in batch_lines}
>         rows = [row for row in rows if row["custom_id"] in wanted]
>     else:
>         input_path = "batch_input.jsonl"
>         batch_lines = en.build_batch_lines(rows)
>         en.write_jsonl(batch_lines, input_path)
>     if not batch_lines:
>         print("Nothing to send.")
>         return 0
>     with open(input_path, "rb") as handle:
>         content_hash = hashlib.sha256(handle.read()).hexdigest()[:16]
>     print(f"{input_path}: {len(batch_lines)} requests, model {batch_lines[0]['body']['model']}")
>
>     state_path = "batch_state.json"
>     state = load_state(state_path) if args.resume else {}
>     if state and state.get("content_hash") != content_hash:
>         print("batch_state.json is for a different input file; drop --resume.")
>         return 2
>     state["content_hash"] = content_hash
>
>     headers = {"Authorization": f"Bearer {api_key}"}
>     try:
>         with httpx.Client(headers=headers, timeout=120.0) as client:
>             if "file_id" not in state:
>                 state["file_id"] = upload_input(client, input_path)
>                 save_state(state_path, state)
>             print(f"Input file: {state['file_id']}")
>
>             if "batch_id" not in state:
>                 state.setdefault("job", args.job)
>                 state.setdefault("idempotency_key", f"{state['job']}-{state['file_id']}")
>                 save_state(state_path, state)  # persist before a create response can be lost
>                 batch = create_batch(client, state["file_id"], state["idempotency_key"], state["job"])
>                 state["batch_id"] = batch["id"]
>                 save_state(state_path, state)
>             print(f"Batch: {state['batch_id']} (polling; safe to Ctrl-C and --resume)")
>
>             batch = poll_batch(client, state["batch_id"])
>             output_lines = download_file(client, batch.get("output_file_id"), "batch_output.jsonl")
>             error_lines = download_file(client, batch.get("error_file_id"), "batch_errors.jsonl")
>     except (BatchError, httpx.HTTPError) as error:
>         print(f"Batch API failed: {error}", file=sys.stderr)
>         return 1
>
>     results = en.collect_results(rows, output_lines, error_lines)
>     en.write_enriched_csv(rows, results, args.out)
>     retries = en.retry_lines(batch_lines, results)
>     en.write_jsonl(retries, "batch_retry.jsonl")
>
>     summary = en.summarize(results)
>     usage = batch.get("usage") or {}
>     final_at = [batch.get(k) for k in ("completed_at", "expired_at", "cancelled_at", "failed_at")]
>     ended_at = next((t for t in final_at if t), None)
>     batch_minutes = (ended_at - batch["created_at"]) / 60 if ended_at else 0.0
>     print(
>         f"\nBatch {batch['id']} ended {batch['status']} after {batch_minutes:.1f} min "
>         f"(this process waited {(time.monotonic() - started) / 60:.1f} min)"
>     )
>     if batch["status"] == "failed":
>         print(f"  batch errors: {json.dumps(batch.get('errors'))}")
>     print(
>         f"  rows: {summary['total']}  completed: {summary['completed']}  "
>         f"needs retry: {summary['needs_retry']}  by status: {summary['by_status']}"
>     )
>     print(
>         f"  tokens (output file only): {usage.get('input_tokens', 0)} in, "
>         f"{usage.get('output_tokens', 0)} out, {usage.get('total_tokens', 0)} total"
>     )
>     print(f"  reported cost: ${summary['cost_usd']:.4f} (includes ${summary['tool_calls_cost_usd']:.4f} for tools)")
>     print(f"  files: {args.out}, batch_output.jsonl, batch_errors.jsonl, batch_retry.jsonl")
>     for result in results.values():
>         if not result.ok:
>             print(f"  retry {result.custom_id}: {result.status}: {result.error}")
>     return 1 if summary["needs_retry"] or batch["status"] != "completed" else 0
>
>
> if __name__ == "__main__":
>     sys.exit(main())
> ```

### Complete run\_batch.py

Save this as `run_batch.py`. It is the four excerpts above in order, nothing else.

> **Full code - run\_batch.py**
>
> ```python run_batch.py theme={null}
> """Enrich companies.csv overnight with the Batch API.
>
> Steps: read the CSV, write batch_input.jsonl, upload it, create a batch, poll
> until a final status, download the output and error files, join by custom_id
> into companies_enriched.csv, write a retry file, and print a summary.
>
> State (file id, batch id, idempotency key) is saved to batch_state.json after
> each step, so `--resume` picks up where a crash or Ctrl-C left off.
> """
>
> from __future__ import annotations
>
> import argparse
> import hashlib
> import json
> import os
> import sys
> import time
> from typing import Any
>
> import enrichment as en
> import httpx
>
> API = "https://api.perplexity.ai"
> FINAL_STATUSES = {"completed", "expired", "cancelled", "failed"}
> POLL_START_SECONDS = 30
> POLL_MAX_SECONDS = 300
>
>
> class BatchError(Exception):
>     """A batch or file endpoint answered with an error."""
>
>
> def check(response: httpx.Response, step: str) -> httpx.Response:
>     """Raise a readable BatchError on any non-2xx answer."""
>     if response.status_code == 404:
>         raise BatchError(
>             f"{step}: 404. The batch or file was not found, belongs to another "
>             "organization, or has expired. See the Batch API guide."
>         )
>     if response.status_code >= 400:
>         raise BatchError(f"{step}: HTTP {response.status_code} {response.text.strip()}")
>     return response
>
>
> def upload_input(client: httpx.Client, path: str) -> str:
>     """POST /v1/files with purpose=batch. The API validates every line first."""
>     with open(path, "rb") as handle:
>         response = client.post(
>             f"{API}/v1/files",
>             data={"purpose": "batch"},
>             files={"file": (os.path.basename(path), handle, "application/jsonl")},
>         )
>     return check(response, "upload").json()["id"]
>
>
> def create_batch(
>     client: httpx.Client, file_id: str, idempotency_key: str, job: str
> ) -> dict[str, Any]:
>     """POST /v1/batches. A retried create with the same Idempotency-Key returns the same batch."""
>     response = client.post(
>         f"{API}/v1/batches",
>         headers={"Idempotency-Key": idempotency_key},
>         json={
>             "input_file_id": file_id,
>             "endpoint": en.BATCH_URL,
>             "completion_window": "24h",
>             "metadata": {"job": job},
>         },
>     )
>     return check(response, "create batch").json()
>
>
> def poll_batch(client: httpx.Client, batch_id: str) -> dict[str, Any]:
>     """GET /v1/batches/{id} with backoff until the status is final."""
>     delay = POLL_START_SECONDS
>     while True:
>         try:
>             response = client.get(f"{API}/v1/batches/{batch_id}")
>             if response.status_code == 429:
>                 wait = float(response.headers.get("Retry-After", delay))
>                 print(f"  rate limited while polling; waiting {wait:.0f}s")
>                 time.sleep(wait)
>                 continue
>             batch = check(response, "poll").json()
>         except httpx.HTTPError as error:  # network blip; keep the overnight job alive
>             print(f"  poll failed ({error}); retrying in {delay}s")
>             time.sleep(delay)
>             continue
>         counts = batch["request_counts"]
>         print(
>             f"  {time.strftime('%H:%M:%S')} {batch['status']}: "
>             f"{counts['completed']} done, {counts['failed']} failed, {counts['total']} total"
>         )
>         if batch["status"] in FINAL_STATUSES:
>             return batch
>         time.sleep(delay)
>         delay = min(int(delay * 1.5), POLL_MAX_SECONDS)
>
>
> def download_file(client: httpx.Client, file_id: str | None, path: str) -> list[dict]:
>     """GET /v1/files/{id}/content to disk, then parse it. No id means no lines."""
>     if not file_id:
>         with open(path, "wb") as handle:
>             handle.write(b"")
>         return []
>     response = client.get(f"{API}/v1/files/{file_id}/content")
>     with open(path, "wb") as handle:
>         handle.write(check(response, f"download {file_id}").content)
>     return en.read_jsonl(path)
>
>
> def load_state(path: str) -> dict[str, Any]:
>     if os.path.exists(path):
>         with open(path, encoding="utf-8") as handle:
>             return json.load(handle)
>     return {}
>
>
> def save_state(path: str, state: dict[str, Any]) -> None:
>     with open(path, "w", encoding="utf-8") as handle:
>         json.dump(state, handle, indent=2)
>
>
> def main() -> int:
>     parser = argparse.ArgumentParser(description=__doc__)
>     parser.add_argument("--csv", default="companies.csv")
>     parser.add_argument("--out", default="companies_enriched.csv")
>     parser.add_argument("--requests", help="upload this JSONL (for example batch_retry.jsonl)")
>     parser.add_argument("--job", default=f"company-enrichment-{time.strftime('%Y-%m-%d')}")
>     parser.add_argument("--resume", action="store_true", help="reuse batch_state.json")
>     args = parser.parse_args()
>
>     api_key = os.environ.get("PERPLEXITY_API_KEY")
>     if not api_key:
>         print("Set PERPLEXITY_API_KEY first.", file=sys.stderr)
>         return 2
>
>     started = time.monotonic()
>     rows = en.read_companies(args.csv)
>     if args.requests:  # a retry run: upload an existing file, keep only its rows
>         input_path = args.requests
>         batch_lines = en.read_jsonl(input_path)
>         wanted = {line["custom_id"] for line in batch_lines}
>         rows = [row for row in rows if row["custom_id"] in wanted]
>     else:
>         input_path = "batch_input.jsonl"
>         batch_lines = en.build_batch_lines(rows)
>         en.write_jsonl(batch_lines, input_path)
>     if not batch_lines:
>         print("Nothing to send.")
>         return 0
>     with open(input_path, "rb") as handle:
>         content_hash = hashlib.sha256(handle.read()).hexdigest()[:16]
>     print(f"{input_path}: {len(batch_lines)} requests, model {batch_lines[0]['body']['model']}")
>
>     state_path = "batch_state.json"
>     state = load_state(state_path) if args.resume else {}
>     if state and state.get("content_hash") != content_hash:
>         print("batch_state.json is for a different input file; drop --resume.")
>         return 2
>     state["content_hash"] = content_hash
>
>     headers = {"Authorization": f"Bearer {api_key}"}
>     try:
>         with httpx.Client(headers=headers, timeout=120.0) as client:
>             if "file_id" not in state:
>                 state["file_id"] = upload_input(client, input_path)
>                 save_state(state_path, state)
>             print(f"Input file: {state['file_id']}")
>
>             if "batch_id" not in state:
>                 state.setdefault("job", args.job)
>                 state.setdefault("idempotency_key", f"{state['job']}-{state['file_id']}")
>                 save_state(state_path, state)  # persist before a create response can be lost
>                 batch = create_batch(client, state["file_id"], state["idempotency_key"], state["job"])
>                 state["batch_id"] = batch["id"]
>                 save_state(state_path, state)
>             print(f"Batch: {state['batch_id']} (polling; safe to Ctrl-C and --resume)")
>
>             batch = poll_batch(client, state["batch_id"])
>             output_lines = download_file(client, batch.get("output_file_id"), "batch_output.jsonl")
>             error_lines = download_file(client, batch.get("error_file_id"), "batch_errors.jsonl")
>     except (BatchError, httpx.HTTPError) as error:
>         print(f"Batch API failed: {error}", file=sys.stderr)
>         return 1
>
>     results = en.collect_results(rows, output_lines, error_lines)
>     en.write_enriched_csv(rows, results, args.out)
>     retries = en.retry_lines(batch_lines, results)
>     en.write_jsonl(retries, "batch_retry.jsonl")
>
>     summary = en.summarize(results)
>     usage = batch.get("usage") or {}
>     final_at = [batch.get(k) for k in ("completed_at", "expired_at", "cancelled_at", "failed_at")]
>     ended_at = next((t for t in final_at if t), None)
>     batch_minutes = (ended_at - batch["created_at"]) / 60 if ended_at else 0.0
>     print(
>         f"\nBatch {batch['id']} ended {batch['status']} after {batch_minutes:.1f} min "
>         f"(this process waited {(time.monotonic() - started) / 60:.1f} min)"
>     )
>     if batch["status"] == "failed":
>         print(f"  batch errors: {json.dumps(batch.get('errors'))}")
>     print(
>         f"  rows: {summary['total']}  completed: {summary['completed']}  "
>         f"needs retry: {summary['needs_retry']}  by status: {summary['by_status']}"
>     )
>     print(
>         f"  tokens (output file only): {usage.get('input_tokens', 0)} in, "
>         f"{usage.get('output_tokens', 0)} out, {usage.get('total_tokens', 0)} total"
>     )
>     print(f"  reported cost: ${summary['cost_usd']:.4f} (includes ${summary['tool_calls_cost_usd']:.4f} for tools)")
>     print(f"  files: {args.out}, batch_output.jsonl, batch_errors.jsonl, batch_retry.jsonl")
>     for result in results.values():
>         if not result.ok:
>             print(f"  retry {result.custom_id}: {result.status}: {result.error}")
>     return 1 if summary["needs_retry"] or batch["status"] != "completed" else 0
>
>
> if __name__ == "__main__":
>     sys.exit(main())
> ```

## Prove it without an API key

`test_enrichment.py` runs the pure functions against fixtures shaped like the Batch API guide's examples. It checks that the request file has no blank lines, unique ids within 64 bytes, `POST`, `/v1/responses`, and one model throughout; that a completed line parses, a fenced one parses, prose and partial objects are rejected, and `incomplete` is not a success; both error-line shapes; that the join keeps every row and marks a row with no line as `missing`; and that the retry file holds exactly the original lines for the rows that did not complete. Save it as `test_enrichment.py`.

> **test\_enrichment.py**
>
> ```python test_enrichment.py theme={null}
> """Offline tests for enrichment.py. No API key, no network."""
>
> from __future__ import annotations
>
> import csv
> import json
>
> import enrichment as en
>
> ROWS = [
>     {"name": "Apple", "website": "https://www.apple.com"},
>     {"name": "Nestlé", "website": "https://www.nestle.com"},
>     {"name": "Toyota", "website": "https://global.toyota"},
> ]
> GOOD_JSON = {
>     "headquarters_city": "Cupertino",
>     "founded_year": 1976,
>     "description": "Apple designs consumer electronics, software, and services.",
>     "latest_news_headline": "Apple reports quarterly results",
>     "latest_news_url": "https://example.com/apple-results",
> }
>
>
> def write_rows(tmp_path, rows=ROWS):
>     path = tmp_path / "companies.csv"
>     with open(path, "w", newline="", encoding="utf-8") as handle:
>         writer = csv.DictWriter(handle, fieldnames=["name", "website"])
>         writer.writeheader()
>         writer.writerows(rows)
>     return str(path)
>
>
> def output_line(custom_id, text, status="completed", cost=0.01):
>     """Build an output-file line in the shape the Batch API guide documents."""
>     return {
>         "id": f"batch_req_{custom_id}",
>         "custom_id": custom_id,
>         "response": {
>             "status_code": 200,
>             "request_id": "01M4RFGZMBKRRZ4QYPDRCZ0CA0",
>             "body": {
>                 "id": f"resp_{custom_id}",
>                 "object": "response",
>                 "status": status,
>                 "model": en.MODEL,
>                 "service_tier": "flex",
>                 "output": [
>                     {"type": "search_results", "results": []},
>                     {
>                         "type": "message",
>                         "role": "assistant",
>                         "status": status,
>                         "content": [{"type": "output_text", "text": text}],
>                     },
>                 ],
>                 "usage": {"input_tokens": 10, "output_tokens": 5, "cost": {"total_cost": cost}},
>             },
>         },
>         "error": None,
>     }
>
>
> # --- JSONL writer ---------------------------------------------------------
>
>
> def test_batch_file_is_valid_jsonl(tmp_path):
>     rows = en.read_companies(write_rows(tmp_path))
>     lines = en.build_batch_lines(rows)
>     path = tmp_path / "batch_input.jsonl"
>     en.write_jsonl(lines, str(path))
>
>     raw = path.read_bytes()
>     assert raw.endswith(b"\n")
>     text_lines = raw.decode("utf-8").split("\n")
>     assert text_lines[-1] == ""  # exactly one trailing newline
>     assert all(line.strip() for line in text_lines[:-1]), "no blank lines"
>     assert "\r" not in raw.decode("utf-8")
>
>     parsed = [json.loads(line) for line in text_lines[:-1]]
>     assert len(parsed) == len(ROWS)
>     ids = [line["custom_id"] for line in parsed]
>     assert len(set(ids)) == len(ids), "custom_id must be unique"
>     assert all(1 <= len(cid.encode("utf-8")) <= 64 for cid in ids)
>     assert {line["method"] for line in parsed} == {"POST"}
>     assert {line["url"] for line in parsed} == {"/v1/responses"}
>     assert {line["body"]["model"] for line in parsed} == {en.MODEL}, "one model per file"
>     assert parsed[0]["body"]["tools"] == [{"type": "web_search"}]
>     assert parsed[0]["body"]["response_format"]["type"] == "json_schema"
>
>
> def test_custom_id_is_safe_for_unicode_and_long_names():
>     long_name = "Ü" * 200
>     cid = en.make_custom_id(7, long_name)
>     assert cid.startswith("row-007-")
>     assert len(cid.encode("utf-8")) <= 64
>     assert en.make_custom_id(2, "Nestlé") == "row-002-nestl"
>     assert en.make_custom_id(3, "!!!") == "row-003-row"
>
>
> # --- output and error line parsing -----------------------------------------
>
>
> def test_parse_completed_output_line():
>     line = output_line("row-001-apple", json.dumps(GOOD_JSON), cost=0.03204)
>     line["response"]["body"]["usage"]["cost"]["tool_calls_cost"] = 0.005
>     result = en.parse_output_line(line)
>     assert result.ok
>     assert result.fields == GOOD_JSON
>     assert result.response_id == "resp_row-001-apple"
>     assert result.cost_usd == 0.03204  # the captured total already includes tools
>     assert result.tool_calls_cost_usd == 0.005
>     summary = en.summarize({result.custom_id: result})
>     assert summary["cost_usd"] == 0.032 and summary["tool_calls_cost_usd"] == 0.005
>
>
> def test_parse_tolerates_markdown_fences_and_rejects_prose():
>     fence = "`" * 3
>     fenced = fence + "json\n" + json.dumps(GOOD_JSON) + "\n" + fence
>     assert en.parse_output_line(output_line("a", fenced)).ok
>     prose = en.parse_output_line(output_line("b", "Apple is based in Cupertino."))
>     assert prose.status == "parse_error"
>     partial = en.parse_output_line(output_line("c", json.dumps({"founded_year": 1976})))
>     assert partial.status == "parse_error"
>
>
> def test_incomplete_response_is_not_ok():
>     result = en.parse_output_line(output_line("a", "", status="incomplete"))
>     assert not result.ok
>     assert result.status == "incomplete"
>
>
> def test_schema_rejects_wrong_types_and_extra_fields():
>     for changes in ({"founded_year": True}, {"founded_year": "1976"}, {"description": []}, {"extra": "x"}):
>         assert en.parse_enrichment(json.dumps({**GOOD_JSON, **changes})) is None
>
>
> def test_existing_enrichment_columns_are_not_duplicated(tmp_path):
>     rows = [{"name": "Apple", "custom_id": "a", "headquarters_city": ""}]
>     results = {"a": en.Result("a", "completed", fields=GOOD_JSON)}
>     path = tmp_path / "enriched.csv"
>     en.write_enriched_csv(rows, results, str(path))
>     with path.open() as handle:
>         columns = next(csv.reader(handle))
>     assert len(columns) == len(set(columns))
>
>
> def test_parse_error_line_api_rejection():
>     # Shape 1 from the guide: error is null, response holds the status and body.
>     line = {
>         "id": "batch_req_01M4RFRF95XTJN5HBN1ASJKVMX",
>         "custom_id": "request-7",
>         "response": {
>             "status_code": 400,
>             "request_id": "01M4RFRF95XTJN5HBN1ASJKVMX",
>             "body": {
>                 "error": {
>                     "message": 'validation failed: model "openai/gpt-nope-9" is not supported',
>                     "type": "invalid_request",
>                     "code": 400,
>                 }
>             },
>         },
>         "error": None,
>     }
>     result = en.parse_error_line(line)
>     assert result.status == "error"
>     assert result.error.startswith("HTTP 400: validation failed")
>
>
> def test_parse_error_line_stopped_run():
>     # Shape 2 from the guide: response is null, error has a code and message.
>     line = {
>         "id": "batch_req_01M4RFJC22KH0N346H7E8XH1DY",
>         "custom_id": "cancel-1",
>         "response": None,
>         "error": {
>             "code": "batch_cancelled",
>             "message": "The batch was cancelled before this request finished.",
>         },
>     }
>     result = en.parse_error_line(line)
>     assert result.status == "error"
>     assert result.error == (
>         "batch_cancelled: The batch was cancelled before this request finished."
>     )
>
>
> # --- join, retry file, summary ----------------------------------------------
>
>
> def test_join_keeps_every_row_and_marks_missing(tmp_path):
>     rows = en.read_companies(write_rows(tmp_path))
>     apple, nestle, toyota = (row["custom_id"] for row in rows)
>     outputs = [
>         output_line(toyota, json.dumps(GOOD_JSON)),  # order differs from the input
>         output_line(apple, json.dumps(GOOD_JSON)),
>     ]
>     errors = [
>         {"custom_id": nestle, "response": None, "error": {"code": "batch_expired", "message": "x"}}
>     ]
>     results = en.collect_results(rows, outputs, errors)
>     assert results[apple].ok and results[toyota].ok
>     assert results[nestle].status == "error"
>
>     # Drop Nestlé's error line: the row must still appear, marked missing.
>     results = en.collect_results(rows, outputs, [])
>     assert results[nestle].status == "missing"
>
>     path = tmp_path / "companies_enriched.csv"
>     en.write_enriched_csv(rows, results, str(path))
>     with open(path, newline="", encoding="utf-8") as handle:
>         written = list(csv.DictReader(handle))
>     assert [row["name"] for row in written] == ["Apple", "Nestlé", "Toyota"]
>     assert written[0]["headquarters_city"] == "Cupertino"
>     assert written[0]["status"] == "completed"
>     assert written[1]["status"] == "missing"
>     assert written[1]["headquarters_city"] == ""
>     assert written[2]["founded_year"] == "1976"
>
>
> def test_retry_file_contains_only_failed_rows(tmp_path):
>     rows = en.read_companies(write_rows(tmp_path))
>     batch_lines = en.build_batch_lines(rows)
>     apple, nestle, toyota = (row["custom_id"] for row in rows)
>     outputs = [
>         output_line(apple, json.dumps(GOOD_JSON)),
>         output_line(toyota, "not json"),
>     ]
>     results = en.collect_results(rows, outputs, [])
>     retries = en.retry_lines(batch_lines, results)
>     assert [line["custom_id"] for line in retries] == [nestle, toyota]
>     assert retries[0] == batch_lines[1]  # the exact original request, ready to upload
>
>     path = tmp_path / "batch_retry.jsonl"
>     en.write_jsonl(retries, str(path))
>     assert en.read_jsonl(str(path)) == retries
>
>     summary = en.summarize(results)
>     assert summary == {
>         "total": 3,
>         "completed": 1,
>         "needs_retry": 2,
>         "by_status": {"completed": 1, "missing": 1, "parse_error": 1},
>         "cost_usd": 0.02,
>         "tool_calls_cost_usd": 0.0,
>     }
> ```

> **Run the tests**
>
> ```bash theme={null}
> python -m pytest -q
> ```

```text theme={null}
...........                                                              [100%]
11 passed
```

### Test the batch lifecycle

Save `test_run_batch.py` next to the other files. It runs the HTTP workflow against a local test double, including a lost create response, a next-day resume with the same idempotency key and metadata, downloads, and a partial failure that exits 1. Run both test files with `python -m pytest -q`.

> **test\_run\_batch.py**
>
> ```python test_run_batch.py theme={null}
> """Offline lifecycle checks: create, resume, download, join, and partial failure."""
>
> import json
>
> import httpx
> import pytest
> import run_batch as rb
> from test_enrichment import GOOD_JSON, output_line
>
>
> @pytest.mark.parametrize("complete", [True, False])
> def test_runner_preserves_idempotency_after_lost_create_response(
>     tmp_path, monkeypatch, complete
> ):
>     monkeypatch.chdir(tmp_path)
>     (tmp_path / "companies.csv").write_text(
>         "name,website\nApple,https://www.apple.com\n"
>     )
>     monkeypatch.setenv("PERPLEXITY_API_KEY", "unused-test-key")
>     monkeypatch.setattr("sys.argv", ["run_batch.py", "--job", "first-day"])
>     calls = []
>     create_bodies = []
>
>     def handler(request):
>         calls.append((request.method, request.url.path))
>         if request.url.path == "/v1/files":
>             return httpx.Response(200, json={"id": "file-1"})
>         if request.url.path == "/v1/batches":
>             create_bodies.append(
>                 (request.headers["Idempotency-Key"], json.loads(request.content))
>             )
>             if len(create_bodies) == 1:
>                 raise httpx.ReadTimeout("create response lost", request=request)
>             return httpx.Response(200, json={"id": "batch-1"})
>         if request.url.path == "/v1/batches/batch-1":
>             return httpx.Response(
>                 200,
>                 json={
>                     "id": "batch-1",
>                     "status": "completed",
>                     "created_at": 100,
>                     "completed_at": 160,
>                     "request_counts": {"total": 1, "completed": 1, "failed": 0},
>                     "output_file_id": "output-1",
>                     "error_file_id": None,
>                 },
>             )
>         if request.url.path == "/v1/files/output-1/content":
>             text = json.dumps(GOOD_JSON) if complete else "not JSON"
>             return httpx.Response(
>                 200, text=json.dumps(output_line("row-001-apple", text)) + "\n"
>             )
>         raise AssertionError(f"unexpected request: {request.method} {request.url.path}")
>
>     real_client = httpx.Client
>     monkeypatch.setattr(
>         rb.httpx,
>         "Client",
>         lambda **kwargs: real_client(transport=httpx.MockTransport(handler), **kwargs),
>     )
>     assert rb.main() == 1
>     saved = json.loads((tmp_path / "batch_state.json").read_text())
>     assert saved["idempotency_key"] == "first-day-file-1"
>     monkeypatch.setattr("sys.argv", ["run_batch.py", "--resume", "--job", "next-day"])
>     assert rb.main() == (0 if complete else 1)
>     assert create_bodies[0] == create_bodies[1]
>     assert calls.count(("POST", "/v1/files")) == 1
>     assert (tmp_path / "companies_enriched.csv").exists()
>     assert (tmp_path / "batch_errors.jsonl").read_text() == ""
>     retry = (tmp_path / "batch_retry.jsonl").read_text()
>     assert bool(retry.strip()) is not complete
> ```

## Spot-check one request first

Before you send a whole file, send one row's body to the synchronous endpoint and read what comes back. The body is identical to a batch line, so this checks the prompt, the schema, and the parser against a real response.

> **One synchronous request**
>
> ```python theme={null}
> import json
> import os
>
> import httpx
>
> import enrichment as en
>
> row = en.read_companies("companies.csv")[3]  # Shopify
> response = httpx.post(
>     "https://api.perplexity.ai/v1/agent",
>     headers={"Authorization": f"Bearer {os.environ['PERPLEXITY_API_KEY']}"},
>     json=en.build_request_body(row),
>     timeout=300,
> )
> response.raise_for_status()
> body = response.json()
> line = {"custom_id": row["custom_id"], "response": {"status_code": response.status_code, "body": body}, "error": None}
> result = en.parse_output_line(line)
> print(result.status, json.dumps(result.fields, indent=2, ensure_ascii=False))
> print("cost", body["usage"]["cost"]["total_cost"], "tier", body.get("service_tier"))
> ```

Observed on October 21, 2026 using the request builder and parser extracted from this page. The Shopify response reported two search calls, 9,237 input tokens, 682 output tokens, `service_tier: default`, and a total cost of $0.06412, including $0.04267 for prompt-cache creation.

```text theme={null}
completed {
  "headquarters_city": "Ottawa",
  "founded_year": 2006,
  "description": "Shopify provides a global commerce platform that enables businesses to create and manage online and physical stores, payments, inventory, marketing, shipping, and other retail operations.",
  "latest_news_headline": "Shopify (SHOP) Launches Open Source Next.js Integration For AI Storefronts",
  "latest_news_url": "https://sg.finance.yahoo.com/news/shopify-shop-launches-open-source-170811172.html"
}
cost 0.06412 tier default
```

A second request for Nestlé, row 6, returned five valid fields, including Vevey, 1866, and a Reuters news URL. It reported 8,806 input tokens, 406 output tokens, two search calls, and a total cost of \$0.05645 on the default tier. These are synchronous results, not batch results. Field validation does not independently verify the claims or the news URLs.

## Run it

From the directory with the six files, with the virtual environment active and the key exported:

> **Submit the batch and wait**
>
> ```bash theme={null}
> python run_batch.py
> ```

The script prints the file id and batch id, then one line per poll. Leave it running, or stop it and come back with `python run_batch.py --resume`. Batches target completion within 24 hours; this is not a deadline. The script exits 1 if any row needs retrying or the batch did not complete, while still saving the available results.

Observed on October 21, 2026 on Linux with Python 3.12, using files extracted from this page. All 12 responses completed, passed field validation, and joined to the correct CSV rows. All reported `service_tier: "flex"`. The batch's `created_at` and `completed_at` were 21 seconds apart; polling and download finished in about 32 seconds. The create response was `validating`, followed by observed `in_progress` and `completed` poll responses. No failed, incomplete, or expired responses appeared, and `error_file_id` was `null`.

```text theme={null}
batch_input.jsonl: 12 requests, model openai/gpt-5.6-sol
Input file: file_01M5GB9DPS6JHE3MHJ087RM6KA
Batch: batch_01M5GB9E7ESHHT818FENVTH8Q3 (polling; safe to Ctrl-C and --resume)
  16:03:39 in_progress: 0 done, 0 failed, 12 total
  16:04:09 completed: 12 done, 0 failed, 12 total

Batch batch_01M5GB9E7ESHHT818FENVTH8Q3 ended completed after 0.3 min (this process waited 0.5 min)
  rows: 12  completed: 12  needs retry: 0  by status: {'completed': 12}
  tokens (output file only): 110226 in, 5026 out, 115252 total
  reported cost: $0.3816 (includes $0.0600 for tools)
  files: companies_enriched.csv, batch_output.jsonl, batch_errors.jsonl, batch_retry.jsonl
```

`companies_enriched.csv` has every input row. A `completed` row has the five new columns filled. Any other row has them blank and the reason in `error`. Open `batch_output.jsonl` when you want the full Response for a row, including the search results the model used; `response_id` in the CSV is the key.

The API returned 12 JSONL records with `custom_id`, `response.status_code: 200`, a full Response under `response.body`, and `error: null`. The code joins by id even when output order differs. When the API returns no error-file id, the script writes an empty local `batch_errors.jsonl` so an older run's errors cannot remain there.

## Retry what failed

`batch_retry.jsonl` is a valid batch input file containing only the requests that did not complete. Send it as its own batch and write the results to a separate CSV so the first join stays intact:

> **Retry run**
>
> ```bash theme={null}
> cp batch_retry.jsonl retry_input.jsonl
> python run_batch.py --requests retry_input.jsonl --out companies_retry.csv
> ```

A retry run overwrites `batch_output.jsonl`, `batch_errors.jsonl`, `batch_retry.jsonl`, and `batch_state.json`. Copy the first run's files aside if you want to keep them. Keep `retry_input.jsonl` unchanged so you can resume with `python run_batch.py --resume --requests retry_input.jsonl --out companies_retry.csv`. Inspect failed rows before sending them again; configuration errors need a fix, while transient failures may only need a retry.

The live 12-row run needed no retry. To test selective retry, a separate local copy omitted Apple's result and replaced Microsoft's response text with invalid JSON. The parser kept 10 completed rows, marked one `missing` and one `parse_error`, and selected only the two original requests. That was a local failure fixture, not a service failure.

Submitting those two requests with the command above created `batch_01M5GBBXT2245TAFJZ06224M5N`. Both completed on `flex` and passed field validation. The batch's timestamps were 14 seconds apart; polling and download finished in about 32 seconds. It used 17,189 input tokens and 868 output tokens, and reported \$0.06094 in total request costs. Its retry file was empty.

## What it costs and how long it takes

There is no separate batch price. Requests run on the `flex` tier where the model supports it. Flex halves model token prices only; tool calls such as `web_search` bill at their normal price. Starting 20 to 24 hours after creation, requests that start or resume run on the default tier so the batch can finish before its 48-hour deadline. Requests that never start are not billed.

The 12-row live run reported $0.38162, including $0.06000 in `tool_calls_cost`. The two-row retry reported $0.06094, including $0.01000 for tools. The combined $0.44256 includes $0.07000 in tool charges; the totals are unchanged because `total_cost` already included them. The batch-level `usage` contained token counts, not a cost total. These are API-reported costs for this run, not a billing-ledger reconciliation or a price guarantee; inspect your own results.

The live batch metadata put `expires_at` 48 hours after `created_at`, and both input and output files had 30-day expiry metadata. A duplicate-id upload returned `400` with `error.type: "invalid_request"`; cancelling the completed batch returned `409` with `error.type: "conflict"`. The run did not exercise a live request error file, active cancellation, or actual deadline expiry. The error-file parsers and failed-result paths are covered by the offline tests, not those successful live runs.

`batch_input.jsonl` can hold up to 50,000 requests and 200 MB. Your organization also has a cap on unfinished batch requests at once, from 50,000 at tier 0 to 1,000,000 at tier 6; a create that would exceed it returns `429` with the count and limit. The polling calls themselves have normal request rate limits, which is why the script backs off.

## Adapt it

* **Your CSV.** Any columns work. The prompt reads `name` and `website`; change `build_request_body` if yours are named differently. Everything else keys on `custom_id`.
* **Your fields.** Edit `ENRICHMENT_FIELDS` and `RESPONSE_SCHEMA` together, then the prompt. The tests that check the join and the parser use `ENRICHMENT_FIELDS`, so they follow.
* **Another model.** Pick a supported model from the [models page](/docs/agent-api/models.md), change `MODEL`, and spot-check the fields you care about before submitting the file.
* **Bigger files.** Nothing changes until 50,000 lines or 200 MB. Past that, split the CSV and run one batch per chunk with a different `--job` name.
* **Unattended runs.** Run `python run_batch.py` under `nohup` or a scheduler, and on failure rerun with `--resume`. The state file is per directory.
* **Audit trail.** Keep `batch_output.jsonl`. Each line's `response.body.id` can be fetched later with `GET /v1/agent/{id}`, and the search results in `output` show which pages the answer came from.

## Troubleshooting

Every Batch API error prints as `Batch API failed: <step>: HTTP <status> <error body>`.

* `404` means the batch or file was not found, belongs to another organization, or has expired. Check the id and credential, then see the [Batch API guide](/docs/agent-api/batch.md).
* `upload: HTTP 400` names the first bad line and why: a blank line, a second model, a `stream: true`, a `function` tool, or a `custom_id` over 64 bytes. Nothing is stored; fix the line and rerun.
* `create batch: HTTP 400` with a message about `Idempotency-Key` means the key was reused with a different request. Inspect the saved state and existing batch before starting a fresh upload. Changing `--job` during `--resume` intentionally keeps the original saved key and metadata.
* `create batch: HTTP 429` is the unfinished-request cap. Wait for earlier batches to finish or send a smaller file.
* `HTTP 402` means your credits are exhausted. A running batch pauses its unstarted requests until you add credits; the deadline does not move.
* `HTTP 403` means your organization has Zero Data Retention, and batch is not available to it.
* A batch that ends `expired` still has an output file. The rows that did not start are in the error file as `batch_expired` and land in the retry file automatically.
* `parse_error` rows are completed responses whose text is not the object the schema describes. Open the row's line in `batch_output.jsonl`; if the text is prose or a partial object, tighten `INSTRUCTIONS`. `incomplete` rows stopped before the object was finished, for example at `max_output_tokens`.
* Input, output, and error files expire 30 days after creation. Download before then; after that, `GET /v1/files/{id}/content` returns `404`.