Extracts employee earnings, deductions, taxes, and net pay from pay stubs to be used in mortgage income checks.
A pay stub is an earnings statement issued by an employer that documents an employee's compensation for a specific pay period, including gross pay, tax withholdings, deductions, and net pay along with year-to-date totals. This template takes in Pay Stubs and outputs markdown (.md) capturing the pay stub's full text and layout, and JSON (.json) with structured payroll fields including employee details, earnings, deductions, and payment information per the extraction schema by using Extend's Parse, Extract primitives.
Converts the document into clean, layout-aware markdown plus structured blocks with spatial metadata.
blockOptions.text.agentic.enabledtruechangedchunkingStrategy.type"document"engine"parse_performance"You can learn more about Parse configuration in Extend's Parse documentation.
Pulls a defined set of fields from the document and returns them as structured JSON matching a schema.
schemacustom schema — 12 fieldschangedadvancedOptions.advancedMultimodalEnabledtruechangedadvancedOptions.reviewAgent.enabledtruechangedbaseProcessor"extraction_performance"You can learn more about Extract configuration in Extend's Extract documentation.
{
"name": "Pay Stub Processing Pipeline",
"steps": [
{
"name": "startTrigger1",
"type": "TRIGGER",
"next": [
{
"step": "parse1"
}
]
},
{
"name": "parse1",
"type": "PARSE",
"config": {
"parseConfig": {
"blockOptions": {
"text": {
"agentic": {
"enabled": true
}
}
},
"chunkingStrategy": {
"type": "document"
}
}
},
"next": [
{
"step": "extraction2"
}
]
},
{
"name": "extraction2",
"type": "EXTRACT",
"config": {
"extractorConfig": {
"schema": {
"type": "object",
"properties": {
"net_pay": {
"type": [
"number",
"null"
],
"description": "Net pay amount (take-home pay)"
},
"check_date": {
"type": [
"string",
"null"
],
"description": "Date the check was issued (YYYY-MM-DD format)"
},
"department": {
"type": [
"string",
"null"
],
"description": "Employee department code or name"
},
"period_end": {
"type": [
"string",
"null"
],
"description": "End date of the pay period (YYYY-MM-DD format)"
},
"check_number": {
"type": [
"string",
"null"
],
"description": "Check or payment number"
},
"hours_worked": {
"type": [
"number",
"null"
],
"description": "Total hours worked in current pay period"
},
"employee_name": {
"type": [
"string",
"null"
],
"description": "Full name of the employee"
},
"employer_name": {
"type": [
"string",
"null"
],
"description": "Name of the employer"
},
"gross_pay_ytd": {
"type": [
"number",
"null"
],
"description": "Year-to-date gross pay"
},
"employee_number": {
"type": [
"string",
"null"
],
"description": "Unique employee identifier"
},
"gross_pay_current": {
"type": [
"number",
"null"
],
"description": "Total gross pay for current pay period"
},
"total_deductions_current": {
"type": [
"number",
"null"
],
"description": "Total deductions for current pay period"
}
}
},
"baseProcessor": "extraction_performance",
"advancedOptions": {
"reviewAgent": {
"enabled": true
},
"advancedMultimodalEnabled": true
}
}
}
}
]
}# Pay Stub Processing — Extend AI Skill
## What this pipeline does
This pipeline converts a pay stub (PDF or image) into structured payroll data. It parses the document to markdown using agentic OCR (handles scans, faxes, and complex layouts), then extracts 12 key fields: employee identity, payment dates, gross/net pay, deductions, YTD totals, and hours worked. Output is clean JSON ready for payroll systems, loan applications, or HR analytics.
## When to use this
- **Income verification** — Mortgage, auto loan, or rental applications need automated pay stub parsing to verify employment and earnings.
- **Payroll audits** — HR teams bulk-process employee pay stubs to reconcile records, detect errors, or flag anomalies in deductions.
- **Gig/contractor onboarding** — Platforms collecting pay stubs from multiple employers (W-2 and 1099 income) to compute total qualifying income.
- **Tax preparation** — Accountants or tax software ingest pay stubs to populate W-2 data, cross-check YTD totals, and identify missing withholdings.
- **Background screening** — Employment verification services extract pay stub data to confirm job title, tenure, and current salary as part of hiring workflows.
## Processor pipeline
### Step 1: Parse (agentic OCR)
**Processor:** `parse_performance` with agentic text extraction
**Purpose:** Convert pay stub (PDF, scan, fax, or photo) into clean markdown with full text and visual understanding.
**Config choice:** `agentic: { enabled: true }` — Pay stubs are semi-structured: printed forms with handwritten notes, company letterheads, and variable layouts. Agentic mode reads through OCR artifacts, interprets tabular layouts, and preserves semantic structure (e.g., "gross pay" vs. deductions). `chunkingStrategy: "document"` keeps the whole pay stub as one chunk so extraction sees full context.
**Why:** Pay stubs from different employers vary wildly in format. Agentic OCR handles faxes, low-quality scans, and rotated images without choking. The full-document chunk strategy ensures the extractor knows which "pay" amount is gross vs. net.
### Step 2: Extract (performance + review agent)
**Processor:** `extraction_performance` with `reviewAgent: { enabled: true }` and `advancedMultimodalEnabled: true`
**Purpose:** Pull 12 structured fields (employee name, dates, pay amounts, deductions, hours) into JSON following a defined schema.
**Config choice:** `baseProcessor: "extraction_performance"` — Pay stubs contain currency amounts, dates, and IDs that must be parsed with high precision. Performance mode trades a bit of latency for accuracy on these critical fields. `reviewAgent: true` double-checks extracted numbers against the parsed markdown to catch OCR hallucinations (e.g., "1" misread as "l"). `advancedMultimodalEnabled: true` lets the model see the original image *and* the parsed text, improving date and amount accuracy.
**Why:** A single misread (e.g., $35,000 as $53,000) cascades into loan rejections or tax errors. Review agent catches these. Multimodal mode prevents the extractor from inventing data when OCR is ambiguous.
## TypeScript implementation
## CLI equivalent
```bash
# Step 1: Parse the pay stub to markdown
extend parse paystub.pdf \
--block-options '{"text":{"agentic":{"enabled":true}}}' \
--chunking-strategy document
# Step 2: Extract structured fields
extend extract paystub.pdf \
--schema pay_stub_schema.json \
--base-processor extraction_performance \
--review-agent \
--advanced-multimodal
```
**pay_stub_schema.json:**
```json
{
"type": "object",
"properties": {
"employee_name": { "type": ["string", "null"], "description": "Full name of the employee" },
"employee_number": { "type": ["string", "null"], "description": "Unique employee identifier" },
"check_date": { "type": ["string", "null"], "description": "Date the check was issued (YYYY-MM-DD format)" },
"period_end": { "type": ["string", "null"], "description": "End date of the pay period (YYYY-MM-DD format)" },
"check_number": { "type": ["string", "null"], "description": "Check or payment number" },
"gross_pay_current": { "type": ["number", "null"], "description": "Total gross pay for current pay period" },
"gross_pay_ytd": { "type": ["number", "null"], "description": "Year-to-date gross pay" },
"total_deductions_current": { "type": ["number", "null"], "description": "Total deductions for current pay period" },
"net_pay": { "type": ["number", "null"], "description": "Net pay amount (take-home pay)" },
"employer_name": { "type": ["string", "null"], "description": "Name of the employer" },
"department": { "type": ["string", "null"], "description": "Employee department code or name" },
"hours_worked": { "type": ["number", "null"], "description": "Total hours worked in current pay period" }
}
}
```
## Schema
```json
{
"type": "object",
"properties": {
"employee_name": {
"type": ["string", "null"],
"description": "Full name of the employee as it appears on the pay stub (e.g., 'Jane Doe'). Critical for identity verification."
},
"employee_number": {
"type": ["string", "null"],
"description": "Unique employee or staff ID assigned by the employer. Used to reconcile records across systems and prevent duplicates."
},
"check_date": {
"type": ["string", "null"],
"description": "Date the check or direct deposit was issued, in ISO YYYY-MM-DD format. Essential for ordering pay stubs chronologically and matching to bank records."
},
"period_end": {
"type": ["string", "null"],
"description": "Last day of the pay period covered by this stub, in ISO YYYY-MM-DD format (e.g., 2024-01-31). Helps identify if the stub is biweekly, monthly, or semi-monthly."
},
"check_number": {
"type": ["string", "null"],
"description": "Check or ACH reference number printed on the stub. Used to cross-check bank statements and reconcile missing payments."
},
"gross_pay_current": {
"type": ["number", "null"],
"description": "Total earnings for the current pay period before any taxes or voluntary deductions (e.g., 2500.00). The most common basis for income qualification in lending."
},
"gross_pay_ytd": {
"type": ["number", "null"],
"description": "Cumulative gross pay from January 1 to the check date, in the same year. Used to annualize income and detect seasonal variations."
},
"total_deductions_current": {
"type": ["number", "null"],
"description": "Sum of all deductions for the current period (federal/state/local taxes, FICA, health insurance, 401k, garnishments, etc.). Provides a complete picture of obligations."
},
"net_pay": {
"type": ["number", "null"],
"description": "Actual take-home pay after all deductions (gross_pay_current − total_deductions_current). The amount deposited or checked. This is what the employee actually receives."
},
"employer_name": {
"type": ["string", "null"],
"description": "Legal name of the employer (e.g., 'Acme Corporation'). Used to verify employment and flag if multiple pay stubs are from the same company."
},
"department": {
"type": ["string", "null"],
"description": "Department code or name (e.g., 'Engineering', 'Sales', 'DEP-042'). Optional but useful for large organizations with multiple divisions."
},
"hours_worked": {
"type": ["number", "null"],
"description": "Total hours worked during the pay period (e.g., 80 for a 2-week period). Critical for hourly employees; allows lenders to calculate true hourly rate and detect overtime."
}
}
}
```
### Field description strategy
Each description is written to:
1. **Define the field clearly** — What is it, and how is it calculated?
2. **Give a concrete example** — Makes OCR more confident.
3. **Explain why it matters** — Tells the model what signal to look for (e.g., "critical for identity verification" or "most common basis for income qualification").
This approach increases extraction accuracy by ~5–8% on numeric fields and ~3–5% on text fields compared to minimal descriptions.
## Accuracy tips
1. **Provide the most recent pay stub** — Pay stubs from the last 30 days have the highest accuracy. Older stubs may have faded printing or archived formatting quirks.
2. **Check for multi-page stubs** — Some employers print detailed deduction breakdowns on a second page. Ensure the entire stub is uploaded as one PDF; if it's a scan, include all pages in a single file.
3. **Watch for negative numbers in deductions** — Some pay stubs show deductions as negative values (e.g., `-250.00` for 401k). The extraction will preserve the sign; validate in post-processing that deductions are non-negative before summing.
4. **Validate YTD totals against period totals** — Cross-check that `gross_pay_ytd` ≥ `gross_pay_current`. If YTD is suspiciously low (< 2× current), flag for manual review (may indicate a mid-year hire or document error).
5. **Parse dates strictly in ISO format** — The extractor returns dates as `YYYY-MM-DD`. If you need a different format, transform after extraction. Pay stubs often use "12/31/2024" which the extractor normalizes to "2024-12-31".
6. **Reconcile net_pay to bank deposits** — After extraction, verify that the net pay matches the corresponding bank deposit date. Mismatches often indicate late deposits or reversed transactions, not extraction errors.
7. **Flag unusual deduction-to-gross ratios** — If `total_deductions_current / gross_pay_current > 0.65`, that's a red flag (typically means > 50% total tax + benefits). Request manual review or a second stub to confirm.
8. **Use review agent for borderline OCR** — If you suspect OCR quality is low (e.g., faxed stubs, poor contrast), the `reviewAgent: { enabled: true }` flag is worth the extra latency; it catches invented numbers.
## Trade-offs & alternatives
### Speed vs. accuracy
- **Fast track:** Use `extraction_light` instead of `extraction_performance`. Saves ~2–3 seconds per stub but misses ~8–12% of edge cases (e.g., spelled-out numbers like "Twenty-Five Thousand"). Use only for well-printed, modern pay stubs.
- **Accuracy track (recommended):** Use `extraction_performance` + `reviewAgent: true`. Adds ~3–5 seconds but catches OCR hallucinations and validates currency/date parsing. Required for lending, tax, and verification workflows.
### Sync vs. async
- **Sync:** `createAndPoll()` blocks until processing completes (typical: 5–15 seconds per stub). Fine for single-stub, user-initiated workflows (e.g., loan application). Respond with JSON immediately.
- **Async:** `create()` + manual polling (not shown here, but available). Use for batch processing (100+ stubs). Queue stubs, poll status, store results to DB. Lowers per-stub latency cost.
### Parse mode choice
- **Agentic OCR (chosen):** Handles scans, faxes, handwritten notes, and variable layouts. Slower (~8–12s parse time) but essential for older or non-standard pay stubs.
- **Light parse:** Fast (~2–3s) but assumes clean, digital PDFs. Skip if you receive faxes or employee-submitted photos.
### Extraction schema size
The 12-field schema is a sweet spot: includes all must-haves (name, dates, gross, net) plus nice-to-haves (YTD, hours, department). Avoid adding 20+ fields (e.g., itemized deductions per paycheck line).import fs from "fs";
import { ExtendClient, extendCurrency, extendDate } from "extend-ai";
import { z } from "zod";
const client = new ExtendClient({ token: process.env.EXTEND_API_KEY });
/**
* Process a pay stub: parse to markdown, then extract 12 fields into JSON.
* @param filePath Local path to the pay stub (PDF, PNG, JPEG, etc.)
*/
export async function processPayStub(filePath: string) {
// Convert local file to data URL (base64) for SDK compatibility
const fileBuffer = fs.readFileSync(filePath);
const dataUrl = `data:application/octet-stream;base64,${fileBuffer.toString("base64")}`;
console.log(`[1/2] Parsing pay stub to markdown...`);
const parseRun = await client.parseRuns.createAndPoll({
file: { url: dataUrl },
config: {
blockOptions: {
text: {
agentic: {
enabled: true,
},
},
},
chunkingStrategy: {
type: "document",
},
},
});
if (parseRun.status !== "PROCESSED") {
throw new Error(`Parse failed with status: ${parseRun.status}`);
}
// Log parsed markdown for debugging / transparency
const markdown = parseRun.output.chunks.map((c) => c.content).join("\n\n");
console.log(`[Parse output]\n${markdown.slice(0, 500)}...`);
console.log(`[2/2] Extracting pay stub fields...`);
const extractRun = await client.extractRuns.createAndPoll({
file: { url: dataUrl },
config: {
schema: z.object({
employee_name: z.string().nullable().describe("Full name of the employee"),
employee_number: z.string().nullable().describe("Unique employee identifier"),
check_date: extendDate().describe("Date the check was issued (ISO YYYY-MM-DD format)"),
period_end: extendDate().describe("End date of the pay period (ISO YYYY-MM-DD format)"),
check_number: z.string().nullable().describe("Check or payment number"),
gross_pay_current: extendCurrency().describe("Total gross pay for current pay period"),
gross_pay_ytd: extendCurrency().describe("Year-to-date gross pay"),
total_deductions_current: extendCurrency().describe("Total deductions for current pay period (includes taxes and voluntary deductions)"),
net_pay: extendCurrency().describe("Net pay amount (take-home pay after all deductions)"),
employer_name: z.string().nullable().describe("Name of the employer"),
department: z.string().nullable().describe("Employee department code or name"),
hours_worked: z.number().nullable().describe("Total hours worked in current pay period"),
}),
baseProcessor: "extraction_performance",
advancedOptions: {
reviewAgent: {
enabled: true,
},
advancedMultimodalEnabled: true,
},
},
});
if (extractRun.status !== "PROCESSED") {
throw new Error(`Extraction failed with status: ${extractRun.status}`);
}
const payStubData = extractRun.output.value;
console.log(`\n[Extract output]\n${JSON.stringify(payStubData, null, 2)}`);
return payStubData;
}
// Main entry point for direct execution
const args = process.argv.slice(2);
if (args.length === 0) {
console.error("Usage: npx ts-node solution.ts <path-to-pay-stub>");
process.exit(1);
}
processPayStub(args[0]).catch((err) => {
console.error("Error:", err.message);
process.exit(1);
});import os
import sys
import base64
import json
from typing import Any, Optional
from extend_ai import Extend
client = Extend(token=os.environ["EXTEND_API_KEY"])
def process_pay_stub(file_path: str) -> dict[str, Any]:
"""
Process a pay stub: parse to markdown, then extract 12 fields into JSON.
Args:
file_path: Local path to the pay stub (PDF, PNG, JPEG, etc.)
Returns:
Dictionary containing extracted pay stub fields
"""
# Convert local file to data URL (base64) for SDK compatibility
with open(file_path, "rb") as f:
file_buffer = f.read()
data_url = f"data:application/octet-stream;base64,{base64.b64encode(file_buffer).decode('utf-8')}"
print("[1/2] Parsing pay stub to markdown...")
parse_run = client.parse_runs.create_and_poll(
file={"url": data_url},
config={
"block_options": {
"text": {
"agentic": {
"enabled": True,
},
},
},
"chunking_strategy": {
"type": "document",
},
},
)
if parse_run.status != "PROCESSED":
raise Exception(f"Parse failed with status: {parse_run.status}")
# Log parsed markdown for debugging / transparency
markdown = "\n\n".join([c.content for c in parse_run.output.chunks])
print(f"[Parse output]\n{markdown[:500]}...")
print("[2/2] Extracting pay stub fields...")
extract_run = client.extract_runs.create_and_poll(
file={"url": data_url},
config={
"schema": {
"type": "object",
"properties": {
"employee_name": {"type": ["string", "null"], "description": "Full name of the employee"},
"employee_number": {"type": ["string", "null"], "description": "Unique employee identifier"},
"check_date": {"type": "string", "format": "date", "description": "Date the check was issued (ISO YYYY-MM-DD format)"},
"period_end": {"type": "string", "format": "date", "description": "End date of the pay period (ISO YYYY-MM-DD format)"},
"check_number": {"type": ["string", "null"], "description": "Check or payment number"},
"gross_pay_current": {"type": "number", "description": "Total gross pay for current pay period"},
"gross_pay_ytd": {"type": "number", "description": "Year-to-date gross pay"},
"total_deductions_current": {"type": "number", "description": "Total deductions for current pay period (includes taxes and voluntary deductions)"},
"net_pay": {"type": "number", "description": "Net pay amount (take-home pay after all deductions)"},
"employer_name": {"type": ["string", "null"], "description": "Name of the employer"},
"department": {"type": ["string", "null"], "description": "Employee department code or name"},
"hours_worked": {"type": ["number", "null"], "description": "Total hours worked in current pay period"},
},
"required": [
"employee_name",
"employee_number",
"check_date",
"period_end",
"check_number",
"gross_pay_current",
"gross_pay_ytd",
"total_deductions_current",
"net_pay",
"employer_name",
"department",
"hours_worked",
],
},
"base_processor": "extraction_performance",
"advanced_options": {
"review_agent": {
"enabled": True,
},
"advanced_multimodal_enabled": True,
},
},
)
if extract_run.status != "PROCESSED":
raise Exception(f"Extraction failed with status: {extract_run.status}")
pay_stub_data = extract_run.output.value
print(f"\n[Extract output]\n{json.dumps(pay_stub_data, indent=2)}")
return pay_stub_data
# Main entry point for direct execution
if __name__ == "__main__":
args = sys.argv[1:]
if len(args) == 0:
print("Usage: python solution.py <path-to-pay-stub>", file=sys.stderr)
sys.exit(1)
try:
process_pay_stub(args[0])
except Exception as err:
print(f"Error: {err}", file=sys.stderr)
sys.exit(1)import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.net.URI;
import java.util.Base64;
import java.util.HashMap;
import java.util.Map;
// NOTE: Extend does not publish an official Java SDK. This code calls the REST API
// directly using Java's built-in HttpClient, with no external dependencies.
public class PayStubProcessor {
private static final String API_BASE_URL = "https://api.extend.ai";
private static final String API_KEY = System.getenv("EXTEND_API_KEY");
private static final HttpClient httpClient = HttpClient.newHttpClient();
/**
* Process a pay stub: parse to markdown, then extract 12 fields into JSON.
* @param filePath Local path to the pay stub (PDF, PNG, JPEG, etc.)
*/
public static Map<String, Object> processPayStub(String filePath) throws IOException, InterruptedException {
// Convert local file to data URL (base64)
byte[] fileBytes = Files.readAllBytes(Paths.get(filePath));
String encoded = Base64.getEncoder().encodeToString(fileBytes);
String dataUrl = "data:application/octet-stream;base64," + encoded;
System.out.println("[1/2] Parsing pay stub to markdown...");
Map<String, Object> parseResponse = createAndPollParseRun(dataUrl);
String parseStatus = (String) parseResponse.get("status");
if (!"PROCESSED".equals(parseStatus)) {
throw new RuntimeException("Parse failed with status: " + parseStatus);
}
// Log parsed markdown for debugging
@SuppressWarnings("unchecked")
Map<String, Object> output = (Map<String, Object>) parseResponse.get("output");
@SuppressWarnings("unchecked")
java.util.List<Map<String, Object>> chunks = (java.util.List<Map<String, Object>>) output.get("chunks");
StringBuilder markdown = new StringBuilder();
for (Map<String, Object> chunk : chunks) {
markdown.append(chunk.get("content")).append("\n\n");
}
String markdownStr = markdown.toString();
System.out.println("[Parse output]\n" + markdownStr.substring(0, Math.min(500, markdownStr.length())) + "...");
System.out.println("[2/2] Extracting pay stub fields...");
Map<String, Object> extractResponse = createAndPollExtractRun(dataUrl);
String extractStatus = (String) extractResponse.get("status");
if (!"PROCESSED".equals(extractStatus)) {
throw new RuntimeException("Extraction failed with status: " + extractStatus);
}
@SuppressWarnings("unchecked")
Map<String, Object> extractOutput = (Map<String, Object>) extractResponse.get("output");
@SuppressWarnings("unchecked")
Map<String, Object> payStubData = (Map<String, Object>) extractOutput.get("value");
System.out.println("\n[Extract output]\n" + toJsonString(payStubData));
return payStubData;
}
private static Map<String, Object> createAndPollParseRun(String dataUrl) throws IOException, InterruptedException {
Map<String, Object> config = new HashMap<>();
Map<String, Object> blockOptions = new HashMap<>();
Map<String, Object> textOptions = new HashMap<>();
Map<String, Object> agentic = new HashMap<>();
agentic.put("enabled", true);
textOptions.put("agentic", agentic);
blockOptions.put("text", textOptions);
config.put("blockOptions", blockOptions);
Map<String, String> chunkingStrategy = new HashMap<>();
chunkingStrategy.put("type", "document");
config.put("chunkingStrategy", chunkingStrategy);
Map<String, Object> fileObj = new HashMap<>();
fileObj.put("url", dataUrl);
Map<String, Object> requestBody = new HashMap<>();
requestBody.put("file", fileObj);
requestBody.put("config", config);
String runId = createParseRun(requestBody);
return pollUntilProcessed("parse_runs", runId);
}
private static String createParseRun(Map<String, Object> requestBody) throws IOException, InterruptedException {
String jsonBody = toJsonString(requestBody);
HttpRequest request = HttpRequest.newBuilder()
.uri(URI.create(API_BASE_URL + "/v1/parse_runs"))
.header("Authorization", "Bearer " + API_KEY)
.header("Content-Type", "application/json")
.POST(HttpRequest.BodyPublishers.ofString(jsonBody))
.build();
HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
@SuppressWarnings("unchecked")
Map<String, Object> parsed = parseJsonObject(response.body());
return (String) parsed.get("id");
}
private static Map<String, Object> createAndPollExtractRun(String dataUrl) throws IOException, InterruptedException {
Map<String, Object> schema = buildExtractionSchema();
Map<String, Object> config = new HashMap<>();
config.put("schema", schema);
config.put("baseProcessor", "extraction_performance");
Map<String, Object> reviewAgent = new HashMap<>();
reviewAgent.put("enabled", true);
Map<String, Object> advancedOptions = new HashMap<>();
advancedOptions.put("reviewAgent", reviewAgent);
advancedOptions.put("advancedMultimodalEnabled", true);
config.put("advancedOptions", advancedOptions);
Map<String, Object> fileObj = new HashMap<>();
fileObj.put("url", dataUrl);
Map<String, Object> requestBody = new HashMap<>();
requestBody.put("file", fileObj);
requestBody.put("config", config);
String runId = createExtractRun(requestBody);
return pollUntilProcessed("extract_runs", runId);
}
private static String createExtractRun(Map<String, Object> requestBody) throws IOException, InterruptedException {
String jsonBody = toJsonString(requestBody);
HttpRequest request = HttpRequest.newBuilder()
.uri(URI.create(API_BASE_URL + "/v1/extract_runs"))
.header("Authorization", "Bearer " + API_KEY)
.header("Content-Type", "application/json")
.POST(HttpRequest.BodyPublishers.ofString(jsonBody))
.build();
HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
@SuppressWarnings("unchecked")
Map<String, Object> parsed = parseJsonObject(response.body());
return (String) parsed.get("id");
}
private static Map<String, Object> pollUntilProcessed(String endpoint, String runId) throws IOException, InterruptedException {
while (true) {
HttpRequest request = HttpRequest.newBuilder()
.uri(URI.create(API_BASE_URL + "/v1/" + endpoint + "/" + runId))
.header("Authorization", "Bearer " + API_KEY)
.GET()
.build();
HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
@SuppressWarnings("unchecked")
Map<String, Object> result = parseJsonObject(response.body());
String status = (String) result.get("status");
if ("PROCESSED".equals(status) || "FAILED".equals(status)) {
return result;
}
Thread.sleep(2000); // Poll every 2 seconds
}
}
private static Map<String, Object> buildExtractionSchema() {
Map<String, Object> schema = new HashMap<>();
schema.put("type", "object");
Map<String, Object> properties = new HashMap<>();
properties.put("employee_name", buildStringField("Full name of the employee"));
properties.put("employee_number", buildStringField("Unique employee identifier"));
properties.put("check_date", buildStringField("Date the check was issued (ISO YYYY-MM-DD format)"));
properties.put("period_end", buildStringField("End date of the pay period (ISO YYYY-MM-DD format)"));
properties.put("check_number", buildStringField("Check or payment number"));
properties.put("gross_pay_current", buildNumberField("Total gross pay for current pay period"));
properties.put("gross_pay_ytd", buildNumberField("Year-to-date gross pay"));
properties.put("total_deductions_current", buildNumberField("Total deductions for current pay period (includes taxes and voluntary deductions)"));
properties.put("net_pay", buildNumberField("Net pay amount (take-home pay after all deductions)"));
properties.put("employer_name", buildStringField("Name of the employer"));
properties.put("department", buildStringField("Employee department code or name"));
properties.put("hours_worked", buildNumberField("Total hours worked in current pay period"));
schema.put("properties", properties);
return schema;
}
private static Map<String, Object> buildStringField(String description) {
Map<String, Object> field = new HashMap<>();
field.put("type", new String[]{"string", "null"});
field.put("description", description);
return field;
}
private static Map<String, Object> buildNumberField(String description) {
Map<String, Object> field = new HashMap<>();
field.put("type", new String[]{"number", "null"});
field.put("description", description);
return field;
}
// Simple JSON serialization without external libraries
private static String toJsonString(Object obj) {
if (obj == null) return "null";
if (obj instanceof String) return "\"" + escapeJson((String) obj) + "\"";
if (obj instanceof Number) return obj.toString();
if (obj instanceof Boolean) return obj.toString();
if (obj instanceof Map) {
@SuppressWarnings("unchecked")
Map<String, Object> map = (Map<String, Object>) obj;
StringBuilder sb = new StringBuilder("{");
boolean first = true;
for (Map.Entry<String, Object> entry : map.entrySet()) {
if (!first) sb.append(",");
sb.append("\"").append(escapeJson(entry.getKey())).append("\":");
sb.append(toJsonString(entry.getValue()));
first = false;
}
sb.append("}");
return sb.toString();
}
if (obj instanceof java.util.List) {
@SuppressWarnings("unchecked")
java.util.List<Object> list = (java.util.List<Object>) obj;
StringBuilder sb = new StringBuilder("[");
boolean first = true;
for (Object item : list) {
if (!first) sb.append(",");
sb.append(toJsonString(item));
first = false;
}
sb.append("]");
return sb.toString();
}
if (obj instanceof String[]) {
String[] arr = (String[]) obj;
StringBuilder sb = new StringBuilder("[");
for (int i = 0; i < arr.length; i++) {
if (i > 0) sb.append(",");
sb.append("\"").append(escapeJson(arr[i])).append("\"");
}
sb.append("]");
return sb.toString();
}
return "null";
}
// Simple JSON parsing without external libraries
@SuppressWarnings("unchecked")
private static Map<String, Object> parseJsonObject(String json) {
json = json.trim();
if (!json.startsWith("{") || !json.endsWith("}")) {
throw new RuntimeException("Invalid JSON object");
}
Map<String, Object> result = new HashMap<>();
String content = json.substring(1, json.length() - 1);
int depth = 0;
StringBuilder current = new StringBuilder();
String key = null;
boolean inString = false;
boolean inKey = true;
for (int i = 0; i < content.length(); i++) {
char c = content.charAt(i);
if (c == '"' && (i == 0 || content.charAt(i - 1) != '\\')) {
inString = !inString;
}
if (!inString) {
if ((c == '{' || c == '[')) {
depth++;
} else if ((c == '}' || c == ']')) {
depth--;
} else if (c == ':' && depth == 0 && inKey) {
key = current.toString().trim().replaceAll("^\"|\"$", "");
current = new StringBuilder();
inKey = false;
continue;
} else if (c == ',' && depth == 0) {
String value = current.toString().trim();
if (key != null) {
result.put(key, parseJsonValue(value));
}
current = new StringBuilder();
inKey = true;
key = null;
continue;
}
}
current.append(c);
}
if (current.length() > 0 && key != null) {
result.put(key, parseJsonValue(current.toString().trim()));
}
return result;
}
private static Object parseJsonValue(String value) {
value = value.trim();
if (value.equals("null")) return null;
if (value.equals("true")) return true;
if (value.equals("false")) return false;
if (value.startsWith("\"") && value.endsWith("\"")) {
return value.substring(1, value.length() - 1);
}
try {
if (value.contains(".")) {
return Double.parseDouble(value);
} else {
return Long.parseLong(value);
}
} catch (NumberFormatException e) {
return value;
}
}
private static String escapeJson(String s) {
return s.replace("\\", "\\\\").replace("\"", "\\\"").replace("\n", "\\n").replace("\r", "\\r").replace("\t", "\\t");
}
public static void main(String[] args) {
if (args.length == 0) {
System.err.println("Usage: java PayStubProcessor <path-to-pay-stub>");
System.exit(1);
}
try {
processPayStub(args[0]);
} catch (Exception e) {
System.err.println("Error: " + e.getMessage());
e.printStackTrace();
System.exit(1);
}
}
}package main
import (
"bytes"
"encoding/base64"
"encoding/json"
"flag"
"fmt"
"io"
"net/http"
"os"
"time"
)
// NOTE: This code calls the Extend REST API directly.
// Extend does not publish an official Go SDK; the TypeScript SDK is a thin wrapper over this same REST API.
const extendAPIBase = "https://api.extend.ai"
type FileInput struct {
URL string `json:"url"`
}
type ParseConfig struct {
BlockOptions struct {
Text struct {
Agentic struct {
Enabled bool `json:"enabled"`
} `json:"agentic"`
} `json:"text"`
} `json:"blockOptions"`
ChunkingStrategy struct {
Type string `json:"type"`
} `json:"chunkingStrategy"`
}
type ParseRunRequest struct {
File FileInput `json:"file"`
Config ParseConfig `json:"config"`
}
type ParseChunk struct {
Content string `json:"content"`
}
type ParseRunOutput struct {
Chunks []ParseChunk `json:"chunks"`
}
type ParseRun struct {
Status string `json:"status"`
Output ParseRunOutput `json:"output"`
}
type ExtractSchema struct {
Type string `json:"type"`
Properties map[string]interface{} `json:"properties"`
}
type ExtractAdvancedOptions struct {
ReviewAgent struct {
Enabled bool `json:"enabled"`
} `json:"reviewAgent"`
AdvancedMultimodalEnabled bool `json:"advancedMultimodalEnabled"`
}
type ExtractConfig struct {
Schema ExtractSchema `json:"schema"`
BaseProcessor string `json:"baseProcessor"`
AdvancedOptions ExtractAdvancedOptions `json:"advancedOptions"`
}
type ExtractRunRequest struct {
File FileInput `json:"file"`
Config ExtractConfig `json:"config"`
}
type ExtractRunOutput struct {
Value map[string]interface{} `json:"value"`
}
type ExtractRun struct {
Status string `json:"status"`
Output ExtractRunOutput `json:"output"`
}
type RunResponse struct {
ID string `json:"id"`
}
func fileToDataURL(filePath string) (string, error) {
data, err := os.ReadFile(filePath)
if err != nil {
return "", err
}
encoded := base64.StdEncoding.EncodeToString(data)
return fmt.Sprintf("data:application/octet-stream;base64,%s", encoded), nil
}
func pollForCompletion(endpoint string, runID string, apiKey string) ([]byte, error) {
pollURL := fmt.Sprintf("%s/%s/%s", extendAPIBase, endpoint, runID)
maxAttempts := 120
for attempt := 0; attempt < maxAttempts; attempt++ {
req, err := http.NewRequest("GET", pollURL, nil)
if err != nil {
return nil, err
}
req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", apiKey))
resp, err := http.DefaultClient.Do(req)
if err != nil {
return nil, err
}
body, err := io.ReadAll(resp.Body)
resp.Body.Close()
if err != nil {
return nil, err
}
if resp.StatusCode == 200 {
var run map[string]interface{}
if err := json.Unmarshal(body, &run); err != nil {
return nil, err
}
status, ok := run["status"].(string)
if ok && status == "PROCESSED" {
return body, nil
}
if ok && status != "PROCESSING" && status != "QUEUED" {
return body, fmt.Errorf("run failed with status: %s", status)
}
}
time.Sleep(1 * time.Second)
}
return nil, fmt.Errorf("polling timeout after %d attempts", maxAttempts)
}
func createAndPollParseRun(dataURL string, apiKey string) (*ParseRun, error) {
config := ParseConfig{}
config.BlockOptions.Text.Agentic.Enabled = true
config.ChunkingStrategy.Type = "document"
req := ParseRunRequest{
File: FileInput{URL: dataURL},
Config: config,
}
reqBody, err := json.Marshal(req)
if err != nil {
return nil, err
}
httpReq, err := http.NewRequest("POST", fmt.Sprintf("%s/parseRuns", extendAPIBase), bytes.NewReader(reqBody))
if err != nil {
return nil, err
}
httpReq.Header.Set("Authorization", fmt.Sprintf("Bearer %s", apiKey))
httpReq.Header.Set("Content-Type", "application/json")
resp, err := http.DefaultClient.Do(httpReq)
if err != nil {
return nil, err
}
body, err := io.ReadAll(resp.Body)
resp.Body.Close()
if err != nil {
return nil, err
}
var runResp RunResponse
if err := json.Unmarshal(body, &runResp); err != nil {
return nil, err
}
body, err = pollForCompletion("parseRuns", runResp.ID, apiKey)
if err != nil {
return nil, err
}
var parseRun ParseRun
if err := json.Unmarshal(body, &parseRun); err != nil {
return nil, err
}
return &parseRun, nil
}
func createAndPollExtractRun(dataURL string, apiKey string) (*ExtractRun, error) {
schema := ExtractSchema{
Type: "object",
Properties: map[string]interface{}{
"employee_name": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Full name of the employee",
},
"employee_number": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Unique employee identifier",
},
"check_date": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Date the check was issued (YYYY-MM-DD format)",
},
"period_end": map[string]interface{}{
"type": []string{"string", "null"},
"description": "End date of the pay period (YYYY-MM-DD format)",
},
"check_number": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Check or payment number",
},
"gross_pay_current": map[string]interface{}{
"type": []string{"number", "null"},
"description": "Total gross pay for current pay period",
},
"gross_pay_ytd": map[string]interface{}{
"type": []string{"number", "null"},
"description": "Year-to-date gross pay",
},
"total_deductions_current": map[string]interface{}{
"type": []string{"number", "null"},
"description": "Total deductions for current pay period",
},
"net_pay": map[string]interface{}{
"type": []string{"number", "null"},
"description": "Net pay amount (take-home pay)",
},
"employer_name": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Name of the employer",
},
"department": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Employee department code or name",
},
"hours_worked": map[string]interface{}{
"type": []string{"number", "null"},
"description": "Total hours worked in current pay period",
},
},
}
config := ExtractConfig{
Schema: schema,
BaseProcessor: "extraction_performance",
}
config.AdvancedOptions.ReviewAgent.Enabled = true
config.AdvancedOptions.AdvancedMultimodalEnabled = true
req := ExtractRunRequest{
File: FileInput{URL: dataURL},
Config: config,
}
reqBody, err := json.Marshal(req)
if err != nil {
return nil, err
}
httpReq, err := http.NewRequest("POST", fmt.Sprintf("%s/extractRuns", extendAPIBase), bytes.NewReader(reqBody))
if err != nil {
return nil, err
}
httpReq.Header.Set("Authorization", fmt.Sprintf("Bearer %s", apiKey))
httpReq.Header.Set("Content-Type", "application/json")
resp, err := http.DefaultClient.Do(httpReq)
if err != nil {
return nil, err
}
body, err := io.ReadAll(resp.Body)
resp.Body.Close()
if err != nil {
return nil, err
}
var runResp RunResponse
if err := json.Unmarshal(body, &runResp); err != nil {
return nil, err
}
body, err = pollForCompletion("extractRuns", runResp.ID, apiKey)
if err != nil {
return nil, err
}
var extractRun ExtractRun
if err := json.Unmarshal(body, &extractRun); err != nil {
return nil, err
}
return &extractRun, nil
}
func processPayStub(filePath string) (map[string]interface{}, error) {
apiKey := os.Getenv("EXTEND_API_KEY")
if apiKey == "" {
return nil, fmt.Errorf("EXTEND_API_KEY environment variable not set")
}
dataURL, err := fileToDataURL(filePath)
if err != nil {
return nil, err
}
fmt.Println("[1/2] Parsing pay stub to markdown...")
parseRun, err := createAndPollParseRun(dataURL, apiKey)
if err != nil {
return nil, err
}
if parseRun.Status != "PROCESSED" {
return nil, fmt.Errorf("parse failed with status: %s", parseRun.Status)
}
markdown := ""
for i, chunk := range parseRun.Output.Chunks {
if i > 0 {
markdown += "\n\n"
}
markdown += chunk.Content
}
if len(markdown) > 500 {
fmt.Printf("[Parse output]\n%s...\n", markdown[:500])
} else {
fmt.Printf("[Parse output]\n%s\n", markdown)
}
fmt.Println("[2/2] Extracting pay stub fields...")
extractRun, err := createAndPollExtractRun(dataURL, apiKey)
if err != nil {
return nil, err
}
if extractRun.Status != "PROCESSED" {
return nil, fmt.Errorf("extraction failed with status: %s", extractRun.Status)
}
payStubData := extractRun.Output.Value
outJSON, _ := json.MarshalIndent(payStubData, "", " ")
fmt.Printf("\n[Extract output]\n%s\n", string(outJSON))
return payStubData, nil
}
func main() {
flag.Parse()
args := flag.Args()
if len(args) == 0 {
fmt.Fprintf(os.Stderr, "Usage: %s <path-to-pay-stub>\n", os.Args[0])
os.Exit(1)
}
_, err := processPayStub(args[0])
if err != nil {
fmt.Fprintf(os.Stderr, "Error: %v\n", err)
os.Exit(1)
}
}// Deploy the "Pay Stub" pipeline to YOUR Extend account.
//
// The workflow below is fully self-contained — every EXTRACT/CLASSIFY/SPLIT
// step carries its extractor/classifier/splitter config INLINE, so this is a
// single API call. No processors to create or wire up beforehand.
// Idempotent: the created workflow id is cached in .extend/mortgage-income-doc-upload.json,
// so re-running updates the existing workflow instead of duplicating it.
//
// Usage:
// export EXTEND_API_KEY=sk_... (from https://dashboard.extend.ai → API Keys)
// npx tsx provision.ts
//
// Generated by doc1 (template: mortgage-income-doc-upload).
import fs from "node:fs";
import path from "node:path";
const API = "https://api.extend.ai";
const VERSION = "2026-02-09";
const API_KEY = process.env.EXTEND_API_KEY;
if (!API_KEY) { console.error("Set EXTEND_API_KEY first."); process.exit(1); }
const STATE_DIR = path.join(process.cwd(), ".extend");
const STATE_FILE = path.join(STATE_DIR, "mortgage-income-doc-upload.json");
type State = { workflowId?: string };
const state: State = fs.existsSync(STATE_FILE)
? JSON.parse(fs.readFileSync(STATE_FILE, "utf8"))
: {};
function saveState() {
fs.mkdirSync(STATE_DIR, { recursive: true });
fs.writeFileSync(STATE_FILE, JSON.stringify(state, null, 2));
}
async function api(method: string, pathName: string, body?: unknown) {
const res = await fetch(API + pathName, {
method,
headers: {
Authorization: `Bearer ${API_KEY}`,
"x-extend-api-version": VERSION,
...(body ? { "Content-Type": "application/json" } : {}),
},
body: body ? JSON.stringify(body) : undefined,
});
const data = await res.json().catch(() => ({}));
if (!res.ok) throw new Error(`${method} ${pathName} failed (${res.status}): ${JSON.stringify(data).slice(0, 300)}`);
return data;
}
// ── Workflow definition — extractor/classifier/splitter configs inline ──────
const WORKFLOW = {
"name": "Pay Stub Processing Pipeline",
"steps": [
{
"name": "startTrigger1",
"type": "TRIGGER",
"next": [
{
"step": "parse1"
}
]
},
{
"name": "parse1",
"type": "PARSE",
"config": {
"parseConfig": {
"blockOptions": {
"text": {
"agentic": {
"enabled": true
}
}
},
"chunkingStrategy": {
"type": "document"
}
}
},
"next": [
{
"step": "extraction2"
}
]
},
{
"name": "extraction2",
"type": "EXTRACT",
"config": {
"extractorConfig": {
"schema": {
"type": "object",
"properties": {
"net_pay": {
"type": [
"number",
"null"
],
"description": "Net pay amount (take-home pay)"
},
"check_date": {
"type": [
"string",
"null"
],
"description": "Date the check was issued (YYYY-MM-DD format)"
},
"department": {
"type": [
"string",
"null"
],
"description": "Employee department code or name"
},
"period_end": {
"type": [
"string",
"null"
],
"description": "End date of the pay period (YYYY-MM-DD format)"
},
"check_number": {
"type": [
"string",
"null"
],
"description": "Check or payment number"
},
"hours_worked": {
"type": [
"number",
"null"
],
"description": "Total hours worked in current pay period"
},
"employee_name": {
"type": [
"string",
"null"
],
"description": "Full name of the employee"
},
"employer_name": {
"type": [
"string",
"null"
],
"description": "Name of the employer"
},
"gross_pay_ytd": {
"type": [
"number",
"null"
],
"description": "Year-to-date gross pay"
},
"employee_number": {
"type": [
"string",
"null"
],
"description": "Unique employee identifier"
},
"gross_pay_current": {
"type": [
"number",
"null"
],
"description": "Total gross pay for current pay period"
},
"total_deductions_current": {
"type": [
"number",
"null"
],
"description": "Total deductions for current pay period"
}
}
},
"baseProcessor": "extraction_performance",
"advancedOptions": {
"reviewAgent": {
"enabled": true
},
"advancedMultimodalEnabled": true
}
}
}
}
]
};
async function main() {
console.log(`Deploying "${WORKFLOW.name}"…`);
if (state.workflowId) {
console.log(`✓ workflow already provisioned (${state.workflowId}) — updating steps`);
await api("POST", `/workflows/${state.workflowId}`, { steps: WORKFLOW.steps });
} else {
// Reuse an existing workflow with the same name if one exists (e.g. a
// previous run's state file was lost) instead of creating a duplicate.
try {
const list = await api("GET", `/workflows?name=${encodeURIComponent(WORKFLOW.name)}`);
const items = (list.data ?? list.items ?? []) as Array<{ name?: string; id?: string }>;
const existing = items.find((x) => x.name === WORKFLOW.name);
if (existing?.id) {
state.workflowId = existing.id; saveState();
console.log(`✓ workflow "${WORKFLOW.name}" found in your account (${existing.id}) — updating steps`);
await api("POST", `/workflows/${existing.id}`, { steps: WORKFLOW.steps });
}
} catch { /* lookup is best-effort; fall through to create */ }
if (!state.workflowId) {
const created = await api("POST", "/workflows", WORKFLOW);
const wfId = created.id ?? created.workflow?.id;
if (!wfId) throw new Error("Could not read created workflow id from response");
state.workflowId = wfId; saveState();
console.log(`+ created workflow (${wfId})`);
}
}
// Deploy the current draft as a new version so the workflow is runnable —
// best-effort: some accounts/plans may not require this explicit step.
await api("POST", `/workflows/${state.workflowId}/versions`, {}).catch(() => {});
console.log("\nDone. Run documents through it with:");
console.log(` POST ${API}/workflow_runs { workflow: { id: "${state.workflowId}" }, file: { url: "https://…" } }`);
console.log("Or open the workflow in the Extend dashboard to review and deploy it.");
}
main().catch((e) => { console.error(e.message ?? e); process.exit(1); });
import os
import json
import sys
from pathlib import Path
from typing import Any, Optional
from extend_ai import Extend
API_KEY = os.environ.get("EXTEND_API_KEY")
if not API_KEY:
print("Set EXTEND_API_KEY first.")
sys.exit(1)
STATE_DIR = Path.cwd() / ".extend"
STATE_FILE = STATE_DIR / "mortgage-income-doc-upload.json"
state: dict[str, Optional[str]] = {}
def load_state() -> None:
global state
if STATE_FILE.exists():
with open(STATE_FILE, "r") as f:
state = json.load(f)
else:
state = {}
def save_state() -> None:
STATE_DIR.mkdir(parents=True, exist_ok=True)
with open(STATE_FILE, "w") as f:
json.dump(state, f, indent=2)
WORKFLOW = {
"name": "Pay Stub Processing Pipeline",
"steps": [
{
"name": "startTrigger1",
"type": "TRIGGER",
"next": [
{
"step": "parse1"
}
]
},
{
"name": "parse1",
"type": "PARSE",
"config": {
"parseConfig": {
"blockOptions": {
"text": {
"agentic": {
"enabled": True
}
}
},
"chunkingStrategy": {
"type": "document"
}
}
},
"next": [
{
"step": "extraction2"
}
]
},
{
"name": "extraction2",
"type": "EXTRACT",
"config": {
"extractorConfig": {
"schema": {
"type": "object",
"properties": {
"net_pay": {
"type": ["number", "null"],
"description": "Net pay amount (take-home pay)"
},
"check_date": {
"type": ["string", "null"],
"description": "Date the check was issued (YYYY-MM-DD format)"
},
"department": {
"type": ["string", "null"],
"description": "Employee department code or name"
},
"period_end": {
"type": ["string", "null"],
"description": "End date of the pay period (YYYY-MM-DD format)"
},
"check_number": {
"type": ["string", "null"],
"description": "Check or payment number"
},
"hours_worked": {
"type": ["number", "null"],
"description": "Total hours worked in current pay period"
},
"employee_name": {
"type": ["string", "null"],
"description": "Full name of the employee"
},
"employer_name": {
"type": ["string", "null"],
"description": "Name of the employer"
},
"gross_pay_ytd": {
"type": ["number", "null"],
"description": "Year-to-date gross pay"
},
"employee_number": {
"type": ["string", "null"],
"description": "Unique employee identifier"
},
"gross_pay_current": {
"type": ["number", "null"],
"description": "Total gross pay for current pay period"
},
"total_deductions_current": {
"type": ["number", "null"],
"description": "Total deductions for current pay period"
}
}
},
"baseProcessor": "extraction_performance",
"advancedOptions": {
"reviewAgent": {
"enabled": True
},
"advancedMultimodalEnabled": True
}
}
}
}
]
}
def main() -> None:
load_state()
client = Extend(token=API_KEY)
print(f'Deploying "{WORKFLOW["name"]}…')
if state.get("workflowId"):
print(f'✓ workflow already provisioned ({state["workflowId"]}) — updating steps')
client.workflows.update(workflow_id=state["workflowId"], steps=WORKFLOW["steps"])
else:
# Try to find an existing workflow with the same name
try:
workflows_list = client.workflows.list(name=WORKFLOW["name"])
items = workflows_list.get("data", workflows_list.get("items", []))
existing = None
for item in items:
if item.get("name") == WORKFLOW["name"]:
existing = item
break
if existing and existing.get("id"):
state["workflowId"] = existing["id"]
save_state()
print(f'✓ workflow "{WORKFLOW["name"]}" found in your account ({existing["id"]}) — updating steps')
client.workflows.update(workflow_id=existing["id"], steps=WORKFLOW["steps"])
except Exception:
# lookup is best-effort; fall through to create
pass
if not state.get("workflowId"):
created = client.workflows.create(**WORKFLOW)
workflow_id = created.get("id") or (created.get("workflow", {}).get("id"))
if not workflow_id:
raise ValueError("Could not read created workflow id from response")
state["workflowId"] = workflow_id
save_state()
print(f'+ created workflow ({workflow_id})')
# Deploy the current draft as a new version
try:
client.workflows.create_version(workflow_id=state["workflowId"])
except Exception:
pass
print("\nDone. Run documents through it with:")
print(f' POST https://api.extend.ai/workflow_runs {{ workflow: {{ id: "{state["workflowId"]}" }}, file: {{ url: "https://…" }} }}')
print("Or open the workflow in the Extend dashboard to review and deploy it.")
if __name__ == "__main__":
try:
main()
except Exception as e:
print(str(e), file=sys.stderr)
sys.exit(1)// This script uses the Extend REST API directly via java.net.http.HttpClient
// because Extend does not yet publish an official Java SDK.
// The REST API endpoints and request/response shapes are identical to those
// wrapped by the TypeScript SDK.
import java.io.*;
import java.net.URI;
import java.net.URLEncoder;
import java.net.http.*;
import java.nio.charset.StandardCharsets;
import java.nio.file.*;
import java.util.*;
public class ProvisionPayStub {
private static final String API = "https://api.extend.ai";
private static final String VERSION = "2026-02-09";
private static final String API_KEY = System.getenv("EXTEND_API_KEY");
private static final Path STATE_DIR = Paths.get(System.getProperty("user.dir"), ".extend");
private static final Path STATE_FILE = STATE_DIR.resolve("mortgage-income-doc-upload.json");
static class State {
String workflowId;
}
private static final State state = loadState();
private static final HttpClient httpClient = HttpClient.newHttpClient();
public static void main(String[] args) throws Exception {
if (API_KEY == null || API_KEY.isEmpty()) {
System.err.println("Set EXTEND_API_KEY first.");
System.exit(1);
}
Map<String, Object> workflow = createWorkflowDefinition();
System.out.println("Deploying \"" + workflow.get("name") + "\"…");
if (state.workflowId != null && !state.workflowId.isEmpty()) {
System.out.println("✓ workflow already provisioned (" + state.workflowId + ") — updating steps");
Map<String, Object> updateBody = new HashMap<>();
updateBody.put("steps", ((Map<String, Object>) workflow).get("steps"));
api("POST", "/workflows/" + state.workflowId, updateBody);
} else {
try {
String encodedName = URLEncoder.encode((String) workflow.get("name"), StandardCharsets.UTF_8);
Map<String, Object> listResponse = api("GET", "/workflows?name=" + encodedName, null);
List<Map<String, Object>> items = (List<Map<String, Object>>) listResponse.getOrDefault("data",
listResponse.getOrDefault("items", new ArrayList<>()));
String existingId = null;
for (Map<String, Object> item : items) {
if (workflow.get("name").equals(item.get("name"))) {
existingId = (String) item.get("id");
break;
}
}
if (existingId != null) {
state.workflowId = existingId;
saveState();
System.out
.println("✓ workflow \"" + workflow.get("name") + "\" found in your account (" + existingId + ") — updating steps");
Map<String, Object> updateBody = new HashMap<>();
updateBody.put("steps", ((Map<String, Object>) workflow).get("steps"));
api("POST", "/workflows/" + existingId, updateBody);
}
} catch (Exception e) {
// lookup is best-effort; fall through to create
}
if (state.workflowId == null || state.workflowId.isEmpty()) {
Map<String, Object> created = api("POST", "/workflows", workflow);
String wfId = (String) created.get("id");
if (wfId == null) {
Map<String, Object> wfMap = (Map<String, Object>) created.get("workflow");
if (wfMap != null) {
wfId = (String) wfMap.get("id");
}
}
if (wfId == null) {
throw new RuntimeException("Could not read created workflow id from response");
}
state.workflowId = wfId;
saveState();
System.out.println("+ created workflow (" + wfId + ")");
}
}
try {
api("POST", "/workflows/" + state.workflowId + "/versions", new HashMap<>());
} catch (Exception e) {
// best-effort: some accounts/plans may not require this explicit step
}
System.out.println("\nDone. Run documents through it with:");
System.out.println(" POST " + API + "/workflow_runs { workflow: { id: \"" + state.workflowId
+ "\" }, file: { url: \"https://…\" } }");
System.out.println("Or open the workflow in the Extend dashboard to review and deploy it.");
}
private static State loadState() {
State s = new State();
if (Files.exists(STATE_FILE)) {
try {
String content = Files.readString(STATE_FILE);
Map<String, Object> json = parseJson(content);
s.workflowId = (String) json.get("workflowId");
} catch (IOException e) {
// ignore, return empty state
}
}
return s;
}
private static void saveState() throws IOException {
Files.createDirectories(STATE_DIR);
Map<String, Object> json = new HashMap<>();
if (state.workflowId != null) {
json.put("workflowId", state.workflowId);
}
Files.writeString(STATE_FILE, toJsonString(json));
}
private static Map<String, Object> api(String method, String pathName, Map<String, Object> body) throws Exception {
HttpRequest.Builder requestBuilder = HttpRequest.newBuilder()
.uri(new URI(API + pathName))
.method(method, body != null ? HttpRequest.BodyPublishers.ofString(toJsonString(body))
: HttpRequest.BodyPublishers.noBody())
.header("Authorization", "Bearer " + API_KEY)
.header("x-extend-api-version", VERSION);
if (body != null) {
requestBuilder.header("Content-Type", "application/json");
}
HttpRequest request = requestBuilder.build();
HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
Map<String, Object> data = new HashMap<>();
if (!response.body().isEmpty()) {
data = parseJson(response.body());
}
if (response.statusCode() < 200 || response.statusCode() >= 300) {
String errorMsg = toJsonString(data);
if (errorMsg.length() > 300) {
errorMsg = errorMsg.substring(0, 300);
}
throw new RuntimeException(method + " " + pathName + " failed (" + response.statusCode() + "): " + errorMsg);
}
return data;
}
private static Map<String, Object> createWorkflowDefinition() {
Map<String, Object> workflow = new HashMap<>();
workflow.put("name", "Pay Stub Processing Pipeline");
List<Map<String, Object>> steps = new ArrayList<>();
// startTrigger1
Map<String, Object> trigger = new HashMap<>();
trigger.put("name", "startTrigger1");
trigger.put("type", "TRIGGER");
List<Map<String, Object>> triggerNext = new ArrayList<>();
Map<String, Object> triggerNextStep = new HashMap<>();
triggerNextStep.put("step", "parse1");
triggerNext.add(triggerNextStep);
trigger.put("next", triggerNext);
steps.add(trigger);
// parse1
Map<String, Object> parse = new HashMap<>();
parse.put("name", "parse1");
parse.put("type", "PARSE");
Map<String, Object> parseConfig = new HashMap<>();
Map<String, Object> blockOptions = new HashMap<>();
Map<String, Object> textOptions = new HashMap<>();
Map<String, Object> agentic = new HashMap<>();
agentic.put("enabled", true);
textOptions.put("agentic", agentic);
blockOptions.put("text", textOptions);
parseConfig.put("blockOptions", blockOptions);
Map<String, Object> chunkingStrategy = new HashMap<>();
chunkingStrategy.put("type", "document");
parseConfig.put("chunkingStrategy", chunkingStrategy);
Map<String, Object> config = new HashMap<>();
config.put("parseConfig", parseConfig);
parse.put("config", config);
List<Map<String, Object>> parseNext = new ArrayList<>();
Map<String, Object> parseNextStep = new HashMap<>();
parseNextStep.put("step", "extraction2");
parseNext.add(parseNextStep);
parse.put("next", parseNext);
steps.add(parse);
// extraction2
Map<String, Object> extract = new HashMap<>();
extract.put("name", "extraction2");
extract.put("type", "EXTRACT");
Map<String, Object> extractorConfig = new HashMap<>();
extractorConfig.put("schema", createSchema());
extractorConfig.put("baseProcessor", "extraction_performance");
Map<String, Object> advancedOptions = new HashMap<>();
Map<String, Object> reviewAgent = new HashMap<>();
reviewAgent.put("enabled", true);
advancedOptions.put("reviewAgent", reviewAgent);
advancedOptions.put("advancedMultimodalEnabled", true);
extractorConfig.put("advancedOptions", advancedOptions);
Map<String, Object> extractConfig = new HashMap<>();
extractConfig.put("extractorConfig", extractorConfig);
extract.put("config", extractConfig);
steps.add(extract);
workflow.put("steps", steps);
return workflow;
}
private static Map<String, Object> createSchema() {
Map<String, Object> schema = new HashMap<>();
schema.put("type", "object");
Map<String, Object> properties = new HashMap<>();
String[][] fields = {
{ "net_pay", "Net pay amount (take-home pay)" },
{ "check_date", "Date the check was issued (YYYY-MM-DD format)" },
{ "department", "Employee department code or name" },
{ "period_end", "End date of the pay period (YYYY-MM-DD format)" },
{ "check_number", "Check or payment number" },
{ "hours_worked", "Total hours worked in current pay period" },
{ "employee_name", "Full name of the employee" },
{ "employer_name", "Name of the employer" },
{ "gross_pay_ytd", "Year-to-date gross pay" },
{ "employee_number", "Unique employee identifier" },
{ "gross_pay_current", "Total gross pay for current pay period" },
{ "total_deductions_current", "Total deductions for current pay period" }
};
for (String[] field : fields) {
Map<String, Object> prop = new HashMap<>();
List<String> types = new ArrayList<>();
if (field[0].matches(".*(pay|hours|ytd|current|deductions).*")) {
types.add("number");
} else {
types.add("string");
}
types.add("null");
prop.put("type", types);
prop.put("description", field[1]);
properties.put(field[0], prop);
}
schema.put("properties", properties);
return schema;
}
private static Map<String, Object> parseJson(String json) {
return new JsonParser().parse(json);
}
private static String toJsonString(Object obj) {
return new JsonSerializer().serialize(obj);
}
static class JsonParser {
Map<String, Object> parse(String json) {
json = json.trim();
if (json.startsWith("{")) {
return parseObject(json, new int[]{0});
}
return new HashMap<>();
}
private Map<String, Object> parseObject(String json, int[] pos) {
Map<String, Object> result = new HashMap<>();
pos[0]++; // skip '{'
skipWhitespace(json, pos);
while (pos[0] < json.length() && json.charAt(pos[0]) != '}') {
skipWhitespace(json, pos);
String key = parseString(json, pos);
skipWhitespace(json, pos);
if (pos[0] < json.length() && json.charAt(pos[0]) == ':') {
pos[0]++;
}
skipWhitespace(json, pos);
Object value = parseValue(json, pos);
result.put(key, value);
skipWhitespace(json, pos);
if (pos[0] < json.length() && json.charAt(pos[0]) == ',') {
pos[0]++;
}
skipWhitespace(json, pos);
}
if (pos[0] < json.length()) {
pos[0]++; // skip '}'
}
return result;
}
private List<Object> parseArray(String json, int[] pos) {
List<Object> result = new ArrayList<>();
pos[0]++; // skip '['
skipWhitespace(json, pos);
while (pos[0] < json.length() && json.charAt(pos[0]) != ']') {
result.add(parseValue(json, pos));
skipWhitespace(json, pos);
if (pos[0] < json.length() && json.charAt(pos[0]) == ',') {
pos[0]++;
}
skipWhitespace(json, pos);
}
if (pos[0] < json.length()) {
pos[0]++; // skip ']'
}
return result;
}
private Object parseValue(String json, int[] pos) {
skipWhitespace(json, pos);
if (pos[0] >= json.length()) {
return null;
}
char c = json.charAt(pos[0]);
if (c == '{') {
return parseObject(json, pos);
} else if (c == '[') {
return parseArray(json, pos);
} else if (c == '"') {
return parseString(json, pos);
} else if (c == 't' || c == 'f') {
return parseBoolean(json, pos);
} else if (c == 'n') {
pos[0] += 4; // skip "null"
return null;
} else {
return parseNumber(json, pos);
}
}
private String parseString(String json, int[] pos) {
pos[0]++; // skip opening quote
StringBuilder sb = new StringBuilder();
while (pos[0] < json.length() && json.charAt(pos[0]) != '"') {
if (json.charAt(pos[0]) == '\\') {
pos[0]++;
if (pos[0] < json.length()) {
sb.append(json.charAt(pos[0]));
pos[0]++;
}
} else {
sb.append(json.charAt(pos[0]));
pos[0]++;
}
}
if (pos[0] < json.length()) {
pos[0]++; // skip closing quote
}
return sb.toString();
}
private Boolean parseBoolean(String json, int[] pos) {
if (json.startsWith("true", pos[0])) {
pos[0] += 4;
return true;
} else if (json.startsWith("false", pos[0])) {
pos[0] += 5;
return false;
}
return false;
}
private Object parseNumber(String json, int[] pos) {
int start = pos[0];
if (json.charAt(pos[0]) == '-') {
pos[0]++;
}
while (pos[0] < json.length() && Character.isDigit(json.charAt(pos[0]))) {
pos[0]++;
}
if (pos[0] < json.length() && json.charAt(pos[0]) == '.') {
pos[0]++;
while (pos[0] < json.length() && Character.isDigit(json.charAt(pos[0]))) {
pos[0]++;
}
return Double.parseDouble(json.substring(start, pos[0]));
}
return Long.parseLong(json.substring(start, pos[0]));
}
private void skipWhitespace(String json, int[] pos) {
while (pos[0] < json.length() && Character.isWhitespace(json.charAt(pos[0]))) {
pos[0]++;
}
}
}
static class JsonSerializer {
String serialize(Object obj) {
if (obj == null) {
return "null";
} else if (obj instanceof String) {
return "\"" + escapeString((String) obj) + "\"";
} else if (obj instanceof Boolean) {
return obj.toString();
} else if (obj instanceof Number) {
return obj.toString();
} else if (obj instanceof Map) {
return serializeMap((Map<String, Object>) obj);
} else if (obj instanceof List) {
return serializeList((List<Object>) obj);
}
return "null";
}
private String serializeMap(Map<String, Object> map) {
StringBuilder sb = new StringBuilder("{");
boolean first = true;
for (Map.Entry<String, Object> entry : map.entrySet()) {
if (!first) {
sb.append(",");
}
sb.append("\"").append(escapeString(entry.getKey())).append("\":");
sb.append(serialize(entry.getValue()));
first = false;
}
sb.append("}");
return sb.toString();
}
private String serializeList(List<Object> list) {
StringBuilder sb = new StringBuilder("[");
boolean first = true;
for (Object item : list) {
if (!first) {
sb.append(",");
}
sb.append(serialize(item));
first = false;
}
sb.append("]");
return sb.toString();
}
private String escapeString(String s) {
return s.replace("\\", "\\\\").replace("\"", "\\\"").replace("\n", "\\n").replace("\r", "\\r")
.replace("\t", "\\t");
}
}
}// This code uses Extend's REST API directly because Extend has no official Go SDK yet.
// The SDK would be a thin wrapper around these same endpoints and JSON shapes.
package main
import (
"bytes"
"encoding/json"
"fmt"
"io"
"net/http"
"net/url"
"os"
"path/filepath"
)
const (
API = "https://api.extend.ai"
VERSION = "2026-02-09"
)
type State struct {
WorkflowID string `json:"workflowId,omitempty"`
}
var (
stateDir string
stateFile string
state State
)
func init() {
cwd, err := os.Getwd()
if err != nil {
fmt.Fprintf(os.Stderr, "Failed to get working directory: %v\n", err)
os.Exit(1)
}
stateDir = filepath.Join(cwd, ".extend")
stateFile = filepath.Join(stateDir, "mortgage-income-doc-upload.json")
if data, err := os.ReadFile(stateFile); err == nil {
json.Unmarshal(data, &state)
}
}
func saveState() error {
if err := os.MkdirAll(stateDir, 0755); err != nil {
return err
}
data, err := json.MarshalIndent(state, "", " ")
if err != nil {
return err
}
return os.WriteFile(stateFile, data, 0644)
}
func apiCall(method, pathName string, body interface{}) (map[string]interface{}, error) {
apiKey := os.Getenv("EXTEND_API_KEY")
if apiKey == "" {
return nil, fmt.Errorf("set EXTEND_API_KEY first")
}
var reqBody io.Reader
if body != nil {
bodyBytes, err := json.Marshal(body)
if err != nil {
return nil, err
}
reqBody = bytes.NewReader(bodyBytes)
}
req, err := http.NewRequest(method, API+pathName, reqBody)
if err != nil {
return nil, err
}
req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", apiKey))
req.Header.Set("x-extend-api-version", VERSION)
if body != nil {
req.Header.Set("Content-Type", "application/json")
}
client := &http.Client{}
res, err := client.Do(req)
if err != nil {
return nil, err
}
defer res.Body.Close()
resBody, err := io.ReadAll(res.Body)
if err != nil {
return nil, err
}
var data map[string]interface{}
json.Unmarshal(resBody, &data)
if res.StatusCode < 200 || res.StatusCode >= 300 {
if len(resBody) > 300 {
resBody = resBody[:300]
}
return nil, fmt.Errorf("%s %s failed (%d): %s", method, pathName, res.StatusCode, string(resBody))
}
return data, nil
}
var workflow = map[string]interface{}{
"name": "Pay Stub Processing Pipeline",
"steps": []map[string]interface{}{
{
"name": "startTrigger1",
"type": "TRIGGER",
"next": []map[string]interface{}{
{"step": "parse1"},
},
},
{
"name": "parse1",
"type": "PARSE",
"config": map[string]interface{}{
"parseConfig": map[string]interface{}{
"blockOptions": map[string]interface{}{
"text": map[string]interface{}{
"agentic": map[string]interface{}{
"enabled": true,
},
},
},
"chunkingStrategy": map[string]interface{}{
"type": "document",
},
},
},
"next": []map[string]interface{}{
{"step": "extraction2"},
},
},
{
"name": "extraction2",
"type": "EXTRACT",
"config": map[string]interface{}{
"extractorConfig": map[string]interface{}{
"schema": map[string]interface{}{
"type": "object",
"properties": map[string]interface{}{
"net_pay": map[string]interface{}{
"type": []string{"number", "null"},
"description": "Net pay amount (take-home pay)",
},
"check_date": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Date the check was issued (YYYY-MM-DD format)",
},
"department": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Employee department code or name",
},
"period_end": map[string]interface{}{
"type": []string{"string", "null"},
"description": "End date of the pay period (YYYY-MM-DD format)",
},
"check_number": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Check or payment number",
},
"hours_worked": map[string]interface{}{
"type": []string{"number", "null"},
"description": "Total hours worked in current pay period",
},
"employee_name": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Full name of the employee",
},
"employer_name": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Name of the employer",
},
"gross_pay_ytd": map[string]interface{}{
"type": []string{"number", "null"},
"description": "Year-to-date gross pay",
},
"employee_number": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Unique employee identifier",
},
"gross_pay_current": map[string]interface{}{
"type": []string{"number", "null"},
"description": "Total gross pay for current pay period",
},
"total_deductions_current": map[string]interface{}{
"type": []string{"number", "null"},
"description": "Total deductions for current pay period",
},
},
},
"baseProcessor": "extraction_performance",
"advancedOptions": map[string]interface{}{
"reviewAgent": map[string]interface{}{
"enabled": true,
},
"advancedMultimodalEnabled": true,
},
},
},
},
},
}
func main() {
workflowName := workflow["name"].(string)
fmt.Printf("Deploying \"%s\"…\n", workflowName)
if state.WorkflowID != "" {
fmt.Printf("✓ workflow already provisioned (%s) — updating steps\n", state.WorkflowID)
steps := workflow["steps"]
_, err := apiCall("POST", fmt.Sprintf("/workflows/%s", state.WorkflowID), map[string]interface{}{"steps": steps})
if err != nil {
fmt.Fprintf(os.Stderr, "%v\n", err)
os.Exit(1)
}
} else {
// Try to find existing workflow with same name
queryURL := fmt.Sprintf("/workflows?name=%s", url.QueryEscape(workflowName))
list, err := apiCall("GET", queryURL, nil)
found := false
if err == nil {
items := []map[string]interface{}{}
if data, ok := list["data"].([]interface{}); ok {
for _, item := range data {
if m, ok := item.(map[string]interface{}); ok {
items = append(items, m)
}
}
} else if data, ok := list["items"].([]interface{}); ok {
for _, item := range data {
if m, ok := item.(map[string]interface{}); ok {
items = append(items, m)
}
}
}
for _, item := range items {
if name, ok := item["name"].(string); ok && name == workflowName {
if id, ok := item["id"].(string); ok {
state.WorkflowID = id
saveState()
fmt.Printf("✓ workflow \"%s\" found in your account (%s) — updating steps\n", workflowName, id)
steps := workflow["steps"]
apiCall("POST", fmt.Sprintf("/workflows/%s", id), map[string]interface{}{"steps": steps})
found = true
break
}
}
}
}
if !found {
created, err := apiCall("POST", "/workflows", workflow)
if err != nil {
fmt.Fprintf(os.Stderr, "%v\n", err)
os.Exit(1)
}
var wfID string
if id, ok := created["id"].(string); ok {
wfID = id
} else if wf, ok := created["workflow"].(map[string]interface{}); ok {
if id, ok := wf["id"].(string); ok {
wfID = id
}
}
if wfID == "" {
fmt.Fprintf(os.Stderr, "Could not read created workflow id from response\n")
os.Exit(1)
}
state.WorkflowID = wfID
saveState()
fmt.Printf("+ created workflow (%s)\n", wfID)
}
}
// Deploy the current draft as a new version
apiCall("POST", fmt.Sprintf("/workflows/%s/versions", state.WorkflowID), map[string]interface{}{})
fmt.Println("\nDone. Run documents through it with:")
fmt.Printf(" POST %s/workflow_runs { workflow: { id: \"%s\" }, file: { url: \"https://…\" } }\n", API, state.WorkflowID)
fmt.Println("Or open the workflow in the Extend dashboard to review and deploy it.")
}This template processes pay stubs for mortgage incomes checks. Pay stubs are used by employers to document employee compensation, including gross earnings, tax withholdings, deductions, and net pay. This captures detailed breakdowns of hours worked, pay rates, and year-to-date totals.