Extracts employee earnings, deductions, taxes, and net pay from pay stubs.
A pay stub is a document issued by an employer that details an employee's compensation for a specific pay period, including gross earnings, tax withholdings, deductions, and net pay, along with year-to-date totals for payroll and tax reporting purposes. This template takes in Pay Stubs and outputs markdown (.md) capturing the pay stub's full text and layout, and JSON (.json) with structured compensation fields including employee information, earnings, deductions, and pay period details 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 json
import base64
from extend_ai import Extend
client = Extend(token=os.environ["EXTEND_API_KEY"])
def process_pay_stub(file_path: str) -> dict:
"""
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={
"blockOptions": {
"text": {
"agentic": {
"enabled": True,
},
},
},
"chunkingStrategy": {
"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", "null"],
"description": "Date the check was issued (ISO YYYY-MM-DD format)",
},
"period_end": {
"type": ["string", "null"],
"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", "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 (includes taxes and voluntary deductions)",
},
"net_pay": {
"type": ["number", "null"],
"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",
},
},
},
"baseProcessor": "extraction_performance",
"advancedOptions": {
"reviewAgent": {
"enabled": True,
},
"advancedMultimodalEnabled": 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
if __name__ == "__main__":
if len(sys.argv) < 2:
print("Usage: python solution.py <path-to-pay-stub>")
sys.exit(1)
try:
process_pay_stub(sys.argv[1])
except Exception as err:
print(f"Error: {err}")
sys.exit(1)// Note: This code calls Extend's REST API directly using Java's built-in HttpClient.
// Extend does not publish an official Java SDK; this approach avoids external dependencies.
import java.io.IOException;
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.Base64;
import java.util.Scanner;
public class PayStubProcessor {
private static final String API_BASE = "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 PayStubData processPayStub(String filePath) throws IOException, InterruptedException {
// Convert local file to data URL (base64)
byte[] fileBuffer = Files.readAllBytes(Paths.get(filePath));
String dataUrl = "data:application/octet-stream;base64," + Base64.getEncoder().encodeToString(fileBuffer);
System.out.println("[1/2] Parsing pay stub to markdown...");
ParseRunResponse parseRun = createAndPollParseRun(dataUrl);
if (!"PROCESSED".equals(parseRun.status)) {
throw new RuntimeException("Parse failed with status: " + parseRun.status);
}
// Log parsed markdown for debugging / transparency
StringBuilder markdown = new StringBuilder();
if (parseRun.output != null && parseRun.output.chunks != null) {
for (Chunk chunk : parseRun.output.chunks) {
markdown.append(chunk.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...");
ExtractRunResponse extractRun = createAndPollExtractRun(dataUrl);
if (!"PROCESSED".equals(extractRun.status)) {
throw new RuntimeException("Extraction failed with status: " + extractRun.status);
}
PayStubData payStubData = extractRun.output.value;
System.out.println("\n[Extract output]\n" + jsonPrettyPrint(payStubData));
return payStubData;
}
private static ParseRunResponse createAndPollParseRun(String dataUrl) throws IOException, InterruptedException {
String parseRequestBody = "{\"file\":{\"url\":\"" + escapeJson(dataUrl) + "\"},\"config\":{\"blockOptions\":{\"text\":{\"agentic\":{\"enabled\":true}}},\"chunkingStrategy\":{\"type\":\"document\"}}}";
HttpRequest createRequest = HttpRequest.newBuilder()
.uri(URI.create(API_BASE + "/v1/parseRuns"))
.header("Authorization", "Bearer " + API_KEY)
.header("Content-Type", "application/json")
.POST(HttpRequest.BodyPublishers.ofString(parseRequestBody))
.build();
HttpResponse<String> createResponse = httpClient.send(createRequest, HttpResponse.BodyHandlers.ofString());
if (createResponse.statusCode() != 200 && createResponse.statusCode() != 201) {
throw new RuntimeException("Parse run creation failed: " + createResponse.body());
}
String responseBody = createResponse.body();
String runId = extractJsonField(responseBody, "id");
return pollParseRun(runId);
}
private static ParseRunResponse pollParseRun(String runId) throws IOException, InterruptedException {
while (true) {
HttpRequest getRequest = HttpRequest.newBuilder()
.uri(URI.create(API_BASE + "/v1/parseRuns/" + runId))
.header("Authorization", "Bearer " + API_KEY)
.GET()
.build();
HttpResponse<String> getResponse = httpClient.send(getRequest, HttpResponse.BodyHandlers.ofString());
if (getResponse.statusCode() != 200) {
throw new RuntimeException("Failed to poll parse run: " + getResponse.body());
}
String responseBody = getResponse.body();
String status = extractJsonField(responseBody, "status");
if ("PROCESSED".equals(status) || "FAILED".equals(status)) {
return parseParseRunResponse(responseBody);
}
Thread.sleep(1000);
}
}
private static ExtractRunResponse createAndPollExtractRun(String dataUrl) throws IOException, InterruptedException {
String schemaJson = buildExtractionSchema();
String extractRequestBody = "{\"file\":{\"url\":\"" + escapeJson(dataUrl) + "\"},\"config\":{\"schema\":" + schemaJson + ",\"baseProcessor\":\"extraction_performance\",\"advancedOptions\":{\"reviewAgent\":{\"enabled\":true},\"advancedMultimodalEnabled\":true}}}";
HttpRequest createRequest = HttpRequest.newBuilder()
.uri(URI.create(API_BASE + "/v1/extractRuns"))
.header("Authorization", "Bearer " + API_KEY)
.header("Content-Type", "application/json")
.POST(HttpRequest.BodyPublishers.ofString(extractRequestBody))
.build();
HttpResponse<String> createResponse = httpClient.send(createRequest, HttpResponse.BodyHandlers.ofString());
if (createResponse.statusCode() != 200 && createResponse.statusCode() != 201) {
throw new RuntimeException("Extract run creation failed: " + createResponse.body());
}
String responseBody = createResponse.body();
String runId = extractJsonField(responseBody, "id");
return pollExtractRun(runId);
}
private static ExtractRunResponse pollExtractRun(String runId) throws IOException, InterruptedException {
while (true) {
HttpRequest getRequest = HttpRequest.newBuilder()
.uri(URI.create(API_BASE + "/v1/extractRuns/" + runId))
.header("Authorization", "Bearer " + API_KEY)
.GET()
.build();
HttpResponse<String> getResponse = httpClient.send(getRequest, HttpResponse.BodyHandlers.ofString());
if (getResponse.statusCode() != 200) {
throw new RuntimeException("Failed to poll extract run: " + getResponse.body());
}
String responseBody = getResponse.body();
String status = extractJsonField(responseBody, "status");
if ("PROCESSED".equals(status) || "FAILED".equals(status)) {
return parseExtractRunResponse(responseBody);
}
Thread.sleep(1000);
}
}
private static String buildExtractionSchema() {
return "{\"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\"}}}";
}
private static ParseRunResponse parseParseRunResponse(String json) {
ParseRunResponse resp = new ParseRunResponse();
resp.status = extractJsonField(json, "status");
resp.output = new ParseOutput();
resp.output.chunks = new Chunk[0];
String chunksStr = extractJsonObject(json, "chunks");
if (chunksStr != null && !chunksStr.isEmpty()) {
resp.output.chunks = new Chunk[1];
resp.output.chunks[0] = new Chunk();
resp.output.chunks[0].content = chunksStr;
}
return resp;
}
private static ExtractRunResponse parseExtractRunResponse(String json) {
ExtractRunResponse resp = new ExtractRunResponse();
resp.status = extractJsonField(json, "status");
resp.output = new ExtractOutput();
resp.output.value = new PayStubData();
String valueStr = extractJsonObject(json, "value");
if (valueStr != null) {
resp.output.value.employee_name = extractJsonField(valueStr, "employee_name");
resp.output.value.employee_number = extractJsonField(valueStr, "employee_number");
resp.output.value.check_date = extractJsonField(valueStr, "check_date");
resp.output.value.period_end = extractJsonField(valueStr, "period_end");
resp.output.value.check_number = extractJsonField(valueStr, "check_number");
resp.output.value.gross_pay_current = extractJsonNumber(valueStr, "gross_pay_current");
resp.output.value.gross_pay_ytd = extractJsonNumber(valueStr, "gross_pay_ytd");
resp.output.value.total_deductions_current = extractJsonNumber(valueStr, "total_deductions_current");
resp.output.value.net_pay = extractJsonNumber(valueStr, "net_pay");
resp.output.value.employer_name = extractJsonField(valueStr, "employer_name");
resp.output.value.department = extractJsonField(valueStr, "department");
resp.output.value.hours_worked = extractJsonNumber(valueStr, "hours_worked");
}
return resp;
}
private static String extractJsonField(String json, String fieldName) {
String searchStr = "\"" + fieldName + "\":";
int idx = json.indexOf(searchStr);
if (idx == -1) return null;
int startIdx = idx + searchStr.length();
while (startIdx < json.length() && Character.isWhitespace(json.charAt(startIdx))) {
startIdx++;
}
if (startIdx >= json.length()) return null;
if (json.charAt(startIdx) == '"') {
StringBuilder result = new StringBuilder();
startIdx++;
while (startIdx < json.length() && json.charAt(startIdx) != '"') {
result.append(json.charAt(startIdx));
startIdx++;
}
return result.toString();
} else if (json.charAt(startIdx) == 'n') {
return null;
}
return null;
}
private static Double extractJsonNumber(String json, String fieldName) {
String searchStr = "\"" + fieldName + "\":";
int idx = json.indexOf(searchStr);
if (idx == -1) return null;
int startIdx = idx + searchStr.length();
while (startIdx < json.length() && Character.isWhitespace(json.charAt(startIdx))) {
startIdx++;
}
if (startIdx >= json.length()) return null;
if (json.charAt(startIdx) == 'n') {
return null;
}
StringBuilder numStr = new StringBuilder();
while (startIdx < json.length() && (Character.isDigit(json.charAt(startIdx)) || json.charAt(startIdx) == '.' || json.charAt(startIdx) == '-')) {
numStr.append(json.charAt(startIdx));
startIdx++;
}
try {
return Double.parseDouble(numStr.toString());
} catch (NumberFormatException e) {
return null;
}
}
private static String extractJsonObject(String json, String fieldName) {
String searchStr = "\"" + fieldName + "\":";
int idx = json.indexOf(searchStr);
if (idx == -1) return null;
int startIdx = idx + searchStr.length();
while (startIdx < json.length() && Character.isWhitespace(json.charAt(startIdx))) {
startIdx++;
}
if (startIdx >= json.length() || json.charAt(startIdx) != '[') return null;
int depth = 0;
int endIdx = startIdx;
while (endIdx < json.length()) {
if (json.charAt(endIdx) == '[') depth++;
else if (json.charAt(endIdx) == ']') depth--;
endIdx++;
if (depth == 0) break;
}
return json.substring(startIdx + 1, endIdx - 1);
}
private static String escapeJson(String str) {
return str.replace("\"", "\\\"").replace("\n", "\\n").replace("\r", "\\r");
}
private static String jsonPrettyPrint(PayStubData data) {
return "{\n \"employee_name\": " + jsonQuote(data.employee_name) + ",\n \"employee_number\": " + jsonQuote(data.employee_number) + ",\n \"check_date\": " + jsonQuote(data.check_date) + ",\n \"period_end\": " + jsonQuote(data.period_end) + ",\n \"check_number\": " + jsonQuote(data.check_number) + ",\n \"gross_pay_current\": " + jsonValue(data.gross_pay_current) + ",\n \"gross_pay_ytd\": " + jsonValue(data.gross_pay_ytd) + ",\n \"total_deductions_current\": " + jsonValue(data.total_deductions_current) + ",\n \"net_pay\": " + jsonValue(data.net_pay) + ",\n \"employer_name\": " + jsonQuote(data.employer_name) + ",\n \"department\": " + jsonQuote(data.department) + ",\n \"hours_worked\": " + jsonValue(data.hours_worked) + "\n}";
}
private static String jsonQuote(String str) {
return str == null ? "null" : "\"" + str.replace("\"", "\\\"") + "\"";
}
private static String jsonValue(Double num) {
return num == null ? "null" : num.toString();
}
public static void main(String[] args) throws IOException, InterruptedException {
if (args.length == 0) {
System.err.println("Usage: java PayStubProcessor <path-to-pay-stub>");
System.exit(1);
}
try {
processPayStub(args[0]);
} catch (Exception err) {
System.err.println("Error: " + err.getMessage());
System.exit(1);
}
}
static class ParseRunResponse {
String status;
ParseOutput output;
}
static class ParseOutput {
Chunk[] chunks;
}
static class Chunk {
String content;
}
static class ExtractRunResponse {
String status;
ExtractOutput output;
}
static class ExtractOutput {
PayStubData value;
}
static class PayStubData {
String employee_name;
String employee_number;
String check_date;
String period_end;
String check_number;
Double gross_pay_current;
Double gross_pay_ytd;
Double total_deductions_current;
Double net_pay;
String employer_name;
String department;
Double hours_worked;
}
}// This code uses the Extend REST API directly because Extend has no official Go SDK yet.
// It makes HTTP calls to https://api.extend.ai with Bearer token authentication.
package main
import (
"bytes"
"encoding/base64"
"encoding/json"
"flag"
"fmt"
"io"
"net/http"
"os"
"time"
)
// PayStubData represents the extracted pay stub fields
type PayStubData struct {
EmployeeName *string `json:"employee_name"`
EmployeeNumber *string `json:"employee_number"`
CheckDate *string `json:"check_date"`
PeriodEnd *string `json:"period_end"`
CheckNumber *string `json:"check_number"`
GrossPayCurrent *float64 `json:"gross_pay_current"`
GrossPayYTD *float64 `json:"gross_pay_ytd"`
TotalDeductionsCurrent *float64 `json:"total_deductions_current"`
NetPay *float64 `json:"net_pay"`
EmployerName *string `json:"employer_name"`
Department *string `json:"department"`
HoursWorked *float64 `json:"hours_worked"`
}
// ParseRunOutput represents the output of a parse run
type ParseRunOutput struct {
Chunks []struct {
Content string `json:"content"`
} `json:"chunks"`
}
// ParseRunResponse represents the response from the parse API
type ParseRunResponse struct {
ID string `json:"id"`
Status string `json:"status"`
Output ParseRunOutput `json:"output"`
Error *string `json:"error"`
}
// ExtractRunResponse represents the response from the extract API
type ExtractRunResponse struct {
ID string `json:"id"`
Status string `json:"status"`
Output struct {
Value PayStubData `json:"value"`
} `json:"output"`
Error *string `json:"error"`
}
// parsePayStub performs the parse step
func parsePayStub(apiKey string, dataURL string) (*ParseRunResponse, error) {
client := &http.Client{Timeout: 300 * time.Second}
payload := map[string]interface{}{
"file": map[string]interface{}{
"url": dataURL,
},
"config": map[string]interface{}{
"blockOptions": map[string]interface{}{
"text": map[string]interface{}{
"agentic": map[string]interface{}{
"enabled": true,
},
},
},
"chunkingStrategy": map[string]interface{}{
"type": "document",
},
},
}
body, err := json.Marshal(payload)
if err != nil {
return nil, fmt.Errorf("failed to marshal parse request: %w", err)
}
req, err := http.NewRequest("POST", "https://api.extend.ai/parseRuns/createAndPoll", bytes.NewBuffer(body))
if err != nil {
return nil, fmt.Errorf("failed to create parse request: %w", err)
}
req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", apiKey))
req.Header.Set("Content-Type", "application/json")
resp, err := client.Do(req)
if err != nil {
return nil, fmt.Errorf("parse request failed: %w", err)
}
defer resp.Body.Close()
respBody, err := io.ReadAll(resp.Body)
if err != nil {
return nil, fmt.Errorf("failed to read parse response: %w", err)
}
var parseResp ParseRunResponse
if err := json.Unmarshal(respBody, &parseResp); err != nil {
return nil, fmt.Errorf("failed to unmarshal parse response: %w", err)
}
return &parseResp, nil
}
// extractPayStub performs the extract step
func extractPayStub(apiKey string, dataURL string) (*ExtractRunResponse, error) {
client := &http.Client{Timeout: 300 * time.Second}
// Build the schema with descriptions
schema := map[string]interface{}{
"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 (ISO YYYY-MM-DD format)",
},
"period_end": map[string]interface{}{
"type": []string{"string", "null"},
"description": "End date of the pay period (ISO 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 (includes taxes and voluntary deductions)",
},
"net_pay": map[string]interface{}{
"type": []string{"number", "null"},
"description": "Net pay amount (take-home pay after all deductions)",
},
"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",
},
},
}
payload := map[string]interface{}{
"file": map[string]interface{}{
"url": dataURL,
},
"config": map[string]interface{}{
"schema": schema,
"baseProcessor": "extraction_performance",
"advancedOptions": map[string]interface{}{
"reviewAgent": map[string]interface{}{
"enabled": true,
},
"advancedMultimodalEnabled": true,
},
},
}
body, err := json.Marshal(payload)
if err != nil {
return nil, fmt.Errorf("failed to marshal extract request: %w", err)
}
req, err := http.NewRequest("POST", "https://api.extend.ai/extractRuns/createAndPoll", bytes.NewBuffer(body))
if err != nil {
return nil, fmt.Errorf("failed to create extract request: %w", err)
}
req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", apiKey))
req.Header.Set("Content-Type", "application/json")
resp, err := client.Do(req)
if err != nil {
return nil, fmt.Errorf("extract request failed: %w", err)
}
defer resp.Body.Close()
respBody, err := io.ReadAll(resp.Body)
if err != nil {
return nil, fmt.Errorf("failed to read extract response: %w", err)
}
var extractResp ExtractRunResponse
if err := json.Unmarshal(respBody, &extractResp); err != nil {
return nil, fmt.Errorf("failed to unmarshal extract response: %w", err)
}
return &extractResp, nil
}
// ProcessPayStub processes a pay stub: parses to markdown, then extracts 12 fields into JSON.
func ProcessPayStub(filePath string) (*PayStubData, error) {
apiKey := os.Getenv("EXTEND_API_KEY")
if apiKey == "" {
return nil, fmt.Errorf("EXTEND_API_KEY environment variable not set")
}
// Read file and convert to data URL
fileBuffer, err := os.ReadFile(filePath)
if err != nil {
return nil, fmt.Errorf("failed to read file: %w", err)
}
dataURL := fmt.Sprintf("data:application/octet-stream;base64,%s", base64.StdEncoding.EncodeToString(fileBuffer))
fmt.Println("[1/2] Parsing pay stub to markdown...")
parseResp, err := parsePayStub(apiKey, dataURL)
if err != nil {
return nil, err
}
if parseResp.Status != "PROCESSED" {
return nil, fmt.Errorf("parse failed with status: %s", parseResp.Status)
}
// Log parsed markdown for debugging / transparency
markdown := ""
for _, chunk := range parseResp.Output.Chunks {
markdown += chunk.Content + "\n\n"
}
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...")
extractResp, err := extractPayStub(apiKey, dataURL)
if err != nil {
return nil, err
}
if extractResp.Status != "PROCESSED" {
return nil, fmt.Errorf("extraction failed with status: %s", extractResp.Status)
}
payStubData := extractResp.Output.Value
jsonData, err := json.MarshalIndent(payStubData, "", " ")
if err == nil {
fmt.Printf("\n[Extract output]\n%s\n", string(jsonData))
}
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/pay-stub.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: pay-stub).
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, "pay-stub.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 json
import os
import sys
from pathlib import Path
from typing import Any, Optional
from extend_ai import Extend
API = "https://api.extend.ai"
VERSION = "2026-02-09"
API_KEY = os.environ.get("EXTEND_API_KEY")
if not API_KEY:
print("Set EXTEND_API_KEY first.", file=sys.stderr)
sys.exit(1)
STATE_DIR = Path.cwd() / ".extend"
STATE_FILE = STATE_DIR / "pay-stub.json"
def load_state() -> dict[str, Any]:
if STATE_FILE.exists():
return json.loads(STATE_FILE.read_text())
return {}
def save_state(state: dict[str, Any]) -> None:
STATE_DIR.mkdir(parents=True, exist_ok=True)
STATE_FILE.write_text(json.dumps(state, 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
}
}
}
}
]
}
async def main() -> None:
client = Extend(token=API_KEY)
state = load_state()
print(f'Deploying "{WORKFLOW["name"]}…"')
if state.get("workflowId"):
workflow_id = state["workflowId"]
print(f"✓ workflow already provisioned ({workflow_id}) — updating steps")
await client.workflows.update(workflow_id, steps=WORKFLOW["steps"])
else:
try:
list_result = await client.workflows.list(name=WORKFLOW["name"])
items = list_result.data if hasattr(list_result, 'data') else (list_result.items if hasattr(list_result, 'items') else [])
existing = next((x for x in items if x.name == WORKFLOW["name"]), None)
if existing and existing.id:
state["workflowId"] = existing.id
save_state(state)
print(f'✓ workflow "{WORKFLOW["name"]}" found in your account ({existing.id}) — updating steps')
await client.workflows.update(existing.id, steps=WORKFLOW["steps"])
except Exception:
pass
if not state.get("workflowId"):
created = await client.workflows.create(**WORKFLOW)
workflow_id = created.id if hasattr(created, 'id') else (created.workflow.id if hasattr(created, 'workflow') else None)
if not workflow_id:
raise ValueError("Could not read created workflow id from response")
state["workflowId"] = workflow_id
save_state(state)
print(f"+ created workflow ({workflow_id})")
workflow_id = state["workflowId"]
try:
await client.workflows.create_version(workflow_id)
except Exception:
pass
print("\nDone. Run documents through it with:")
print(f' POST {API}/workflow_runs {{ workflow: {{ id: "{workflow_id}" }}, file: {{ url: "https://…" }} }}')
print("Or open the workflow in the Extend dashboard to review and deploy it.")
if __name__ == "__main__":
import asyncio
try:
asyncio.run(main())
except Exception as e:
print(str(e), file=sys.stderr)
sys.exit(1)// NOTE: Extend does not publish an official Java SDK. This code calls the
// REST API directly using only java.net.http.HttpClient (no third-party deps).
import java.io.IOException;
import java.net.URI;
import java.net.URLEncoder;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
public class PayStubProvisioning {
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("pay-stub.json");
static class State {
String workflowId;
State(String workflowId) {
this.workflowId = workflowId;
}
}
private static State state = new State(null);
private static void loadState() throws IOException {
if (Files.exists(STATE_FILE)) {
String json = Files.readString(STATE_FILE);
Map<String, Object> parsed = parseJson(json);
state.workflowId = (String) parsed.get("workflowId");
}
}
private static void saveState() throws IOException {
Files.createDirectories(STATE_DIR);
Map<String, Object> data = new HashMap<>();
data.put("workflowId", state.workflowId);
Files.writeString(STATE_FILE, toJsonString(data, 2));
}
private static Map<String, Object> api(String method, String pathName, Object body) throws IOException, InterruptedException {
java.net.http.HttpClient client = java.net.http.HttpClient.newHttpClient();
java.net.http.HttpRequest.Builder reqBuilder = java.net.http.HttpRequest.newBuilder()
.uri(URI.create(API + pathName))
.header("Authorization", "Bearer " + API_KEY)
.header("x-extend-api-version", VERSION);
if (body != null) {
String jsonBody = toJsonString(body, 0);
reqBuilder.header("Content-Type", "application/json")
.method(method, java.net.http.HttpRequest.BodyPublishers.ofString(jsonBody));
} else {
reqBuilder.method(method, java.net.http.HttpRequest.BodyPublishers.noBody());
}
java.net.http.HttpResponse<String> res = client.send(reqBuilder.build(),
java.net.http.HttpResponse.BodyHandlers.ofString());
Map<String, Object> data;
try {
data = parseJson(res.body());
} catch (Exception e) {
data = new HashMap<>();
}
if (res.statusCode() < 200 || res.statusCode() >= 300) {
String msg = toJsonString(data, 0);
if (msg.length() > 300) msg = msg.substring(0, 300);
throw new RuntimeException(method + " " + pathName + " failed (" + res.statusCode() + "): " + msg);
}
return data;
}
private static Map<String, Object> buildWorkflow() {
Map<String, Object> schema = new LinkedHashMap<>();
schema.put("type", "object");
Map<String, Object> properties = new LinkedHashMap<>();
properties.put("net_pay", Map.of("type", List.of("number", "null"), "description", "Net pay amount (take-home pay)"));
properties.put("check_date", Map.of("type", List.of("string", "null"), "description", "Date the check was issued (YYYY-MM-DD format)"));
properties.put("department", Map.of("type", List.of("string", "null"), "description", "Employee department code or name"));
properties.put("period_end", Map.of("type", List.of("string", "null"), "description", "End date of the pay period (YYYY-MM-DD format)"));
properties.put("check_number", Map.of("type", List.of("string", "null"), "description", "Check or payment number"));
properties.put("hours_worked", Map.of("type", List.of("number", "null"), "description", "Total hours worked in current pay period"));
properties.put("employee_name", Map.of("type", List.of("string", "null"), "description", "Full name of the employee"));
properties.put("employer_name", Map.of("type", List.of("string", "null"), "description", "Name of the employer"));
properties.put("gross_pay_ytd", Map.of("type", List.of("number", "null"), "description", "Year-to-date gross pay"));
properties.put("employee_number", Map.of("type", List.of("string", "null"), "description", "Unique employee identifier"));
properties.put("gross_pay_current", Map.of("type", List.of("number", "null"), "description", "Total gross pay for current pay period"));
properties.put("total_deductions_current", Map.of("type", List.of("number", "null"), "description", "Total deductions for current pay period"));
schema.put("properties", properties);
Map<String, Object> extractorConfig = new LinkedHashMap<>();
extractorConfig.put("schema", schema);
extractorConfig.put("baseProcessor", "extraction_performance");
extractorConfig.put("advancedOptions", Map.of(
"reviewAgent", Map.of("enabled", true),
"advancedMultimodalEnabled", true
));
Map<String, Object> parseConfig = Map.of(
"blockOptions", Map.of(
"text", Map.of("agentic", Map.of("enabled", true))
),
"chunkingStrategy", Map.of("type", "document")
);
List<Map<String, Object>> steps = List.of(
Map.of(
"name", "startTrigger1",
"type", "TRIGGER",
"next", List.of(Map.of("step", "parse1"))
),
Map.of(
"name", "parse1",
"type", "PARSE",
"config", Map.of("parseConfig", parseConfig),
"next", List.of(Map.of("step", "extraction2"))
),
Map.of(
"name", "extraction2",
"type", "EXTRACT",
"config", Map.of("extractorConfig", extractorConfig)
)
);
Map<String, Object> workflow = new LinkedHashMap<>();
workflow.put("name", "Pay Stub Processing Pipeline");
workflow.put("steps", steps);
return workflow;
}
public static void main(String[] args) {
try {
if (API_KEY == null || API_KEY.isEmpty()) {
System.err.println("Set EXTEND_API_KEY first.");
System.exit(1);
}
loadState();
Map<String, Object> workflow = buildWorkflow();
String workflowName = (String) workflow.get("name");
System.out.println("Deploying \"" + workflowName + "\"…");
if (state.workflowId != null) {
System.out.println("✓ workflow already provisioned (" + state.workflowId + ") — updating steps");
api("POST", "/workflows/" + state.workflowId, Map.of("steps", workflow.get("steps")));
} else {
try {
String encoded = URLEncoder.encode(workflowName, StandardCharsets.UTF_8);
Map<String, Object> list = api("GET", "/workflows?name=" + encoded, null);
List<?> items = (List<?>) list.getOrDefault("data", list.getOrDefault("items", List.of()));
Optional<?> existing = items.stream()
.filter(x -> x instanceof Map)
.map(x -> (Map<?, ?>) x)
.filter(x -> workflowName.equals(x.get("name")))
.findFirst();
if (existing.isPresent()) {
Map<?, ?> item = (Map<?, ?>) existing.get();
String id = (String) item.get("id");
if (id != null) {
state.workflowId = id;
saveState();
System.out.println("✓ workflow \"" + workflowName + "\" found in your account (" + id + ") — updating steps");
api("POST", "/workflows/" + id, Map.of("steps", workflow.get("steps")));
}
}
} catch (Exception e) {
// lookup is best-effort; fall through to create
}
if (state.workflowId == null) {
Map<String, Object> created = api("POST", "/workflows", workflow);
String wfId = (String) created.get("id");
if (wfId == null && created.get("workflow") instanceof Map) {
wfId = (String) ((Map<?, ?>) created.get("workflow")).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", Map.of());
} catch (Exception e) {
// best-effort
}
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.");
} catch (Exception e) {
String msg = e.getMessage() != null ? e.getMessage() : e.toString();
System.err.println(msg);
System.exit(1);
}
}
private static Map<String, Object> parseJson(String json) {
json = json.trim();
if (!json.startsWith("{")) return new HashMap<>();
return parseJsonObject(json, new int[]{0});
}
private static Map<String, Object> parseJsonObject(String json, int[] pos) {
Map<String, Object> map = new LinkedHashMap<>();
skipWhitespace(json, pos);
pos[0]++; // skip '{'
while (pos[0] < json.length()) {
skipWhitespace(json, pos);
if (json.charAt(pos[0]) == '}') {
pos[0]++;
break;
}
if (json.charAt(pos[0]) == ',') {
pos[0]++;
skipWhitespace(json, pos);
}
String key = parseJsonString(json, pos);
skipWhitespace(json, pos);
pos[0]++; // skip ':'
Object value = parseJsonValue(json, pos);
map.put(key, value);
}
return map;
}
private static List<?> parseJsonArray(String json, int[] pos) {
List<Object> list = new java.util.ArrayList<>();
skipWhitespace(json, pos);
pos[0]++; // skip '['
while (pos[0] < json.length()) {
skipWhitespace(json, pos);
if (json.charAt(pos[0]) == ']') {
pos[0]++;
break;
}
if (json.charAt(pos[0]) == ',') {
pos[0]++;
skipWhitespace(json, pos);
}
list.add(parseJsonValue(json, pos));
}
return list;
}
private static Object parseJsonValue(String json, int[] pos) {
skipWhitespace(json, pos);
char c = json.charAt(pos[0]);
if (c == '"') return parseJsonString(json, pos);
if (c == '{') return parseJsonObject(json, pos);
if (c == '[') return parseJsonArray(json, pos);
if (c == 't' || c == 'f') {
boolean val = json.startsWith("true", pos[0]);
pos[0] += val ? 4 : 5;
return val;
}
if (c == 'n') {
pos[0] += 4;
return null;
}
int start = pos[0];
while (pos[0] < json.length() && "0123456789.-+eE".indexOf(json.charAt(pos[0])) >= 0) {
pos[0]++;
}
String num = json.substring(start, pos[0]);
return num.contains(".") || num.contains("e") || num.contains("E") ? Double.parseDouble(num) : Long.parseLong(num);
}
private static String parseJsonString(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]++;
char escaped = json.charAt(pos[0]);
sb.append(escaped == 'n' ? '\n' : escaped == 't' ? '\t' : escaped == 'r' ? '\r' : escaped);
} else {
sb.append(json.charAt(pos[0]));
}
pos[0]++;
}
pos[0]++; // skip closing quote
return sb.toString();
}
private static void skipWhitespace(String json, int[] pos) {
while (pos[0] < json.length() && " \t\n\r".indexOf(json.charAt(pos[0])) >= 0) {
pos[0]++;
}
}
private static String toJsonString(Object obj, int indent) {
if (obj == null) return "null";
if (obj instanceof String) return "\"" + escapeJson((String) obj) + "\"";
if (obj instanceof Boolean || obj instanceof Number) return obj.toString();
if (obj instanceof List) {
List<?> list = (List<?>) obj;
StringBuilder sb = new StringBuilder("[");
for (int i = 0; i < list.size(); i++) {
if (i > 0) sb.append(",");
if (indent > 0) sb.append("\n").append(" ".repeat(indent + 2));
sb.append(toJsonString(list.get(i), indent > 0 ? indent + 2 : 0));
}
if (indent > 0 && !list.isEmpty()) sb.append("\n").append(" ".repeat(indent));
sb.append("]");
return sb.toString();
}
if (obj instanceof Map) {
Map<?, ?> map = (Map<?, ?>) obj;
StringBuilder sb = new StringBuilder("{");
boolean first = true;
for (Map.Entry<?, ?> e : map.entrySet()) {
if (!first) sb.append(",");
first = false;
if (indent > 0) sb.append("\n").append(" ".repeat(indent + 2));
sb.append("\"").append(escapeJson(e.getKey().toString())).append("\":");
if (indent > 0) sb.append(" ");
sb.append(toJsonString(e.getValue(), indent > 0 ? indent + 2 : 0));
}
if (indent > 0 && !map.isEmpty()) sb.append("\n").append(" ".repeat(indent));
sb.append("}");
return sb.toString();
}
return "\"" + obj.toString() + "\"";
}
private static String escapeJson(String s) {
return s.replace("\\", "\\\\")
.replace("\"", "\\\"")
.replace("\n", "\\n")
.replace("\r", "\\r")
.replace("\t", "\\t");
}
}// This code calls Extend's REST API directly because Extend has no official Go SDK yet.
// The API is fully self-contained — every EXTRACT/CLASSIFY/SPLIT step carries its
// config INLINE, so this is a single API call. No processors to create beforehand.
// Idempotent: the created workflow id is cached in .extend/pay-stub.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)
// go run provision.go
//
// Generated by doc1 (template: pay-stub).
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"
)
var (
apiKey string
stateDir string
stateFile string
state map[string]interface{}
)
type WorkflowStep struct {
Name string `json:"name"`
Type string `json:"type"`
Next []map[string]string `json:"next,omitempty"`
Config map[string]interface{} `json:"config,omitempty"`
}
type Workflow struct {
Name string `json:"name"`
Steps []WorkflowStep `json:"steps"`
}
type APIResponse struct {
ID string `json:"id,omitempty"`
Workflow map[string]interface{} `json:"workflow,omitempty"`
Data []map[string]interface{} `json:"data,omitempty"`
Items []map[string]interface{} `json:"items,omitempty"`
}
func init() {
apiKey = os.Getenv("EXTEND_API_KEY")
if apiKey == "" {
fmt.Fprintln(os.Stderr, "Set EXTEND_API_KEY first.")
os.Exit(1)
}
stateDir = filepath.Join(".", ".extend")
stateFile = filepath.Join(stateDir, "pay-stub.json")
state = make(map[string]interface{})
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) {
var reqBody io.Reader
if body != nil {
b, err := json.Marshal(body)
if err != nil {
return nil, err
}
reqBody = bytes.NewReader(b)
}
req, err := http.NewRequest(method, API+pathName, reqBody)
if err != nil {
return nil, err
}
req.Header.Set("Authorization", "Bearer "+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()
respBody, err := io.ReadAll(res.Body)
if err != nil {
return nil, err
}
var data map[string]interface{}
json.Unmarshal(respBody, &data)
if res.StatusCode < 200 || res.StatusCode >= 300 {
msg := string(respBody)
if len(msg) > 300 {
msg = msg[:300]
}
return nil, fmt.Errorf("%s %s failed (%d): %s", method, pathName, res.StatusCode, msg)
}
return data, nil
}
func buildWorkflow() Workflow {
return Workflow{
Name: "Pay Stub Processing Pipeline",
Steps: []WorkflowStep{
{
Name: "startTrigger1",
Type: "TRIGGER",
Next: []map[string]string{
{"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]string{
{"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() {
workflow := buildWorkflow()
fmt.Printf("Deploying \"%s\"…\n", workflow.Name)
workflowID, ok := state["workflowId"].(string)
if ok && workflowID != "" {
fmt.Printf("✓ workflow already provisioned (%s) — updating steps\n", workflowID)
_, err := apiCall("POST", "/workflows/"+workflowID, map[string]interface{}{"steps": workflow.Steps})
if err != nil {
fmt.Fprintf(os.Stderr, "%v\n", err)
os.Exit(1)
}
} else {
// Try to find existing workflow by name.
q := url.QueryEscape(workflow.Name)
list, err := apiCall("GET", "/workflows?name="+q, nil)
found := false
if err == nil {
var items []map[string]interface{}
if data, ok := list["data"].([]interface{}); ok {
for _, v := range data {
if m, ok := v.(map[string]interface{}); ok {
items = append(items, m)
}
}
} else if itemsRaw, ok := list["items"].([]interface{}); ok {
for _, v := range itemsRaw {
if m, ok := v.(map[string]interface{}); ok {
items = append(items, m)
}
}
}
for _, item := range items {
if name, ok := item["name"].(string); ok && name == workflow.Name {
if id, ok := item["id"].(string); ok {
workflowID = id
state["workflowId"] = id
saveState()
fmt.Printf("✓ workflow \"%s\" found in your account (%s) — updating steps\n", workflow.Name, id)
_, err := apiCall("POST", "/workflows/"+id, map[string]interface{}{"steps": workflow.Steps})
if err != nil {
fmt.Fprintf(os.Stderr, "%v\n", err)
os.Exit(1)
}
found = true
break
}
}
}
}
if !found {
created, err := apiCall("POST", "/workflows", workflow)
if err != nil {
fmt.Fprintf(os.Stderr, "%v\n", err)
os.Exit(1)
}
wfID := ""
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)
workflowID = wfID
}
}
// Deploy the current draft as a new version — best-effort.
apiCall("POST", "/workflows/"+workflowID+"/versions", 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, workflowID)
fmt.Println("Or open the workflow in the Extend dashboard to review and deploy it.")
}Pay stubs used by employers to document employee compensation, including gross earnings, tax withholdings, deductions, and net pay. This template captures detailed breakdowns of hours worked, pay rates, and year-to-date totals for payroll and tax reporting purposes.