Extracts pension award details, beneficiary info, and benefit amounts from official letters.
A pension award letter is an official document issued by a pension administrator that notifies an employee of their approved retirement benefits, including their personal details, pension plan information, monthly benefit amount, commencement date, and payment terms. This template takes in Pension Award Letter and outputs markdown (.md) preserving the document's full text and structure, and JSON (.json) with extracted pension benefit fields including recipient details, employer information, monthly pension amount, commencement date, and payment frequency 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": "Pension Award Letter 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": {
"letter_date": {
"type": [
"string",
"null"
],
"description": "Date the pension award letter was issued"
},
"contact_email": {
"type": [
"string",
"null"
],
"description": "Email address for inquiries or assistance"
},
"employer_name": {
"type": [
"string",
"null"
],
"description": "Name of the employer/organization issuing the pension"
},
"recipient_name": {
"type": [
"string",
"null"
],
"description": "Full name of the pension award recipient"
},
"signatory_name": {
"type": [
"string",
"null"
],
"description": "Name and title of the person signing the letter"
},
"years_of_service": {
"type": [
"string",
"null"
],
"description": "Total years of service with the employer"
},
"commencement_date": {
"type": [
"string",
"null"
],
"description": "Date when pension payments commence"
},
"payment_frequency": {
"type": [
"string",
"null"
],
"description": "Frequency of pension payments (e.g., Monthly, Quarterly)"
},
"pension_plan_name": {
"type": [
"string",
"null"
],
"description": "Name of the pension plan under which the award is granted"
},
"recipient_address": {
"type": [
"string",
"null"
],
"description": "Complete mailing address of the recipient"
},
"issuing_department": {
"type": [
"string",
"null"
],
"description": "Department or office issuing the pension award letter"
},
"monthly_pension_amount": {
"type": [
"string",
"null"
],
"description": "Monthly pension payment amount in currency"
}
}
},
"baseProcessor": "extraction_performance",
"advancedOptions": {
"reviewAgent": {
"enabled": true
},
"advancedMultimodalEnabled": true
}
}
}
}
]
}# Pension Award Letter Processing — Extend AI Skill
## What this pipeline does
Extracts structured pension benefit information from official award letters issued by government or corporate pension administrators. The pipeline parses the letter to markdown using agentic OCR (to handle varied layouts and formatting), then extracts 12 key fields including recipient details, employer information, pension amounts, commencement dates, and contact information into validated JSON.
## When to use this
- **Pension administration systems**: Auto-populate employee records from incoming award letters to reduce manual data entry and eligibility errors.
- **Retirement planning platforms**: Ingest pension letters to surface benefit amounts, payment dates, and plan details in user dashboards.
- **Benefits compliance & audit**: Extract and archive pension award data with confidence scores for regulatory reporting and reconciliation.
- **Document triage & routing**: Parse letters to route them to appropriate teams (payroll, finance, HR) based on extracted employer and department information.
- **Multi-format sources**: Handle scanned PDFs, faxes, and handwritten annotations alongside clean digital letters using agentic OCR.
## Processor pipeline
**Step 1: Parse (`agentic_ocr` + document-level chunking)**
- **Purpose**: Convert the award letter (scanned, faxed, or digital) to clean markdown while preserving semantic structure.
- **Config choice**: `blockOptions.text.agentic.enabled = true` — Agentic OCR is required because pension letters often have variable layouts, letterhead blocks, signature areas, and multi-column tables that demand intelligent text recovery.
- **Chunking**: `chunkingStrategy.type = "document"` — Treat the entire letter as one chunk since award letters are typically 1–2 pages and coherence across sections (e.g., linking recipient name to amounts) is critical.
- **Why**: Pension award letters mix formal tables, contact info blocks, and certification statements. Agentic OCR recovers text from low-quality scans and preserves section order, which improves extraction accuracy downstream.
**Step 2: Extract (`extraction_performance` + Zod schema + review agent + multimodal)**
- **Purpose**: Pull 12 structured fields (name, address, dates, amounts, contact info, signatory) into validated JSON.
- **Config choice**: `baseProcessor: "extraction_performance"` — Pension letters demand high accuracy for financial and legal fields (amounts, dates, signatory authority). Performance mode uses heavier models and is worth the latency for compliance.
- **Review agent**: `advancedOptions.reviewAgent.enabled = true` — Pension data triggers compliance checks and requires auditability. The review agent flags uncertain extractions (e.g., ambiguous dates, missing signatories) so a human can verify before the letter is filed.
- **Multimodal**: `advancedMultimodalEnabled = true` — Allows the extractor to cross-reference parsed text with visual layouts (e.g., confirming that a box labeled "Monthly Benefit" contains the pension amount, not a deduction).
- **Why**: Pension amounts and commencement dates are legally binding; extraction errors can delay benefit payments or cause regulatory issues. Performance mode + review agent ensures confidence before downstream use.
## TypeScript implementation
```typescript
import { ExtendClient } from "extend-ai";
import { z } from "zod";
import fs from "fs";
const client = new ExtendClient({ token: process.env.EXTEND_API_KEY });
// Zod schema for pension award letter extraction
const pensionAwardSchema = z.object({
recipient_name: z
.string()
.nullable()
.describe(
"Full name of the pension award recipient as printed on the letter"
),
recipient_address: z
.string()
.nullable()
.describe("Complete mailing address of the recipient including postal code"),
letter_date: z
.string()
.nullable()
.describe("Date the pension award letter was issued (ISO format or natural date)"),
pension_plan_name: z
.string()
.nullable()
.describe(
"Name of the pension plan under which the award is granted (e.g., 'Company Pension Plan 2024')"
),
employer_name: z
.string()
.nullable()
.describe(
"Name of the employer or organization issuing the pension (full legal name)"
),
years_of_service: z
.string()
.nullable()
.describe("Total years of service with the employer (numeric or text, e.g., '25 years')"),
monthly_pension_amount: z
.string()
.nullable()
.describe(
"Monthly pension payment amount in currency format (e.g., '$2,345.67' or '€1,500.00')"
),
commencement_date: z
.string()
.nullable()
.describe(
"Date when pension payments commence or have commenced (ISO format or natural date)"
),
payment_frequency: z
.string()
.nullable()
.describe(
"Frequency of pension payments (e.g., 'Monthly', 'Quarterly', 'Annually')"
),
issuing_department: z
.string()
.nullable()
.describe(
"Department or office issuing the pension award letter (e.g., 'Pension Administration Department')"
),
contact_email: z
.string()
.nullable()
.describe(
"Email address for inquiries or assistance regarding the pension award"
),
signatory_name: z
.string()
.nullable()
.describe(
"Name and title of the person signing the letter (e.g., 'Jane Smith, Director of Benefits')"
),
});
export async function processPensionAwardLetter(filePath: string) {
console.log(`Processing pension award letter: ${filePath}`);
// Convert local file to base64 data URL for SDK
const fileBuffer = fs.readFileSync(filePath);
const base64 = fileBuffer.toString("base64");
const dataUrl = `data:application/octet-stream;base64,${base64}`;
try {
// Step 1: Parse the pension award letter using agentic OCR
console.log("Step 1: Parsing pension award letter...");
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}`);
}
const markdown = parseRun.output.chunks
.map((chunk) => chunk.content)
.join("\n\n");
console.log("Parsed markdown (first 500 chars):");
console.log(markdown.substring(0, 500));
console.log("\n");
// Step 2: Extract structured fields from the pension award letter
console.log("Step 2: Extracting structured pension award fields...");
const extractRun = await client.extractRuns.createAndPoll({
file: { url: dataUrl },
config: {
schema: pensionAwardSchema,
baseProcessor: "extraction_performance",
advancedOptions: {
reviewAgent: {
enabled: true,
},
advancedMultimodalEnabled: true,
},
},
});
if (extractRun.status !== "PROCESSED") {
throw new Error(`Extraction failed with status: ${extractRun.status}`);
}
const extractedData = extractRun.output.value;
console.log("Extracted pension award data:");
console.log(JSON.stringify(extractedData, null, 2));
return {
success: true,
parseStatus: parseRun.status,
markdownPreview: markdown.substring(0, 500),
extractedFields: extractedData,
confidence: extractRun.confidence || "N/A",
};
} catch (error) {
console.error("Error processing pension award letter:", error);
throw error;
}
}
// Run the pipeline if this file is executed directly
const filePath = process.argv[2] || "./pension_award_letter.pdf";
processPensionAwardLetter(filePath).then((result) => {
console.log("\nPipeline completed successfully.");
console.log(JSON.stringify(result, null, 2));
});
```
## CLI equivalent
```bash
# Step 1: Parse the pension award letter
extend parse pension_award_letter.pdf
# Step 2: Extract structured fields using the Zod schema
extend extract pension_award_letter.pdf \
--schema pension_award_schema.json \
--base-processor extraction_performance \
--review-agent \
--advanced-multimodal
```
**Schema file** (`pension_award_schema.json`):
```json
{
"recipient_name": {
"type": ["string", "null"],
"description": "Full name of the pension award recipient as printed on the letter"
},
"recipient_address": {
"type": ["string", "null"],
"description": "Complete mailing address of the recipient including postal code"
},
"letter_date": {
"type": ["string", "null"],
"description": "Date the pension award letter was issued (ISO format or natural date)"
},
"pension_plan_name": {
"type": ["string", "null"],
"description": "Name of the pension plan under which the award is granted (e.g., 'Company Pension Plan 2024')"
},
"employer_name": {
"type": ["string", "null"],
"description": "Name of the employer or organization issuing the pension (full legal name)"
},
"years_of_service": {
"type": ["string", "null"],
"description": "Total years of service with the employer (numeric or text, e.g., '25 years')"
},
"monthly_pension_amount": {
"type": ["string", "null"],
"description": "Monthly pension payment amount in currency format (e.g., '$2,345.67' or '€1,500.00')"
},
"commencement_date": {
"type": ["string", "null"],
"description": "Date when pension payments commence or have commenced (ISO format or natural date)"
},
"payment_frequency": {
"type": ["string", "null"],
"description": "Frequency of pension payments (e.g., 'Monthly', 'Quarterly', 'Annually')"
},
"issuing_department": {
"type": ["string", "null"],
"description": "Department or office issuing the pension award letter (e.g., 'Pension Administration Department')"
},
"contact_email": {
"type": ["string", "null"],
"description": "Email address for inquiries or assistance regarding the pension award"
},
"signatory_name": {
"type": ["string", "null"],
"description": "Name and title of the person signing the letter (e.g., 'Jane Smith, Director of Benefits')"
}
}
```
## Schema
The extraction schema is structured to capture the 12 critical fields in a pension award letter:
```json
{
"type": "object",
"properties": {
"recipient_name": {
"type": ["string", "null"],
"description": "Full name of the pension award recipient as printed on the letter. Critical for identity verification and matching against payroll systems."
},
"recipient_address": {
"type": ["string", "null"],
"description": "Complete mailing address of the recipient including postal code. Used for correspondence and regulatory filing; must include street, city, state, and ZIP."
},
"letter_date": {
"type": ["string", "null"],
"description": "Date the pension award letter was issued (ISO format or natural date). Establishes the effective date of the award and triggers benefit commencement windows."
},
"pension_plan_name": {
"type": ["string", "null"],
"description": "Name of the pension plan under which the award is granted (e.g., 'Company Pension Plan 2024', 'Municipal Defined Benefit Plan'). Links the award to the correct plan rules and vesting schedules."
},
"employer_name": {
"type": ["string", "null"],
"description": "Name of the employer or organization issuing the pension (full legal name). Critical for routing letters to correct departments and matching against company records."
},
"years_of_service": {
"type": ["string", "null"],
"description": "Total years of service with the employer (numeric or text, e.g., '25 years', '25'). Determines eligibility and benefit multipliers in pension formulas."
},
"monthly_pension_amount": {
"type": ["string", "null"],
"description": "Monthly pension payment amount in currency format (e.g., '$2,345.67', '€1,500.00'). The core benefit amount; used for budgeting, payroll setup, and tax withholding calculations."
},
"commencement_date": {
"type": ["string", "null"],
"description": "Date when pension payments commence or have commenced (ISO format or natural date). Triggers payment system activation and retroactive benefit calculations."
},
"payment_frequency": {
"type": ["string", "null"],
"description": "Frequency of pension payments (e.g., 'Monthly', 'Quarterly', 'Annually'). Determines payment schedule in accounting and benefits platforms."
},
"issuing_department": {
"type": ["string", "null"],
"description": "Department or office issuing the pension award letter (e.g., 'Pension Administration Department', 'Benefits & Compensation'). Routes questions and appeals to the right team."
},
"contact_email": {
"type": ["string", "null"],
"description": "Email address for inquiries or assistance regarding the pension award. Used to populate support contacts in employee records and for automated notifications."
},
"signatory_name": {
"type": ["string", "null"],
"description": "Name and title of the person signing the letter (e.g., 'Jane Smith, Director of Benefits'). Validates authority of the letter issuer and provides audit trail for compliance."
}
}
}
```
**Field description strategy**: Each description explicitly links the field to its downstream use case (payroll, compliance, routing, tax). This specificity improves extractor accuracy by giving the model context for what "correct" looks like in your system.
## Accuracy tips
1. **Agentic OCR is non-negotiable**: Pension letters arrive from dozens of issuers with different templates, letterheads, and sometimes poor scan quality. Always enable `blockOptions.text.agentic.enabled = true` to recover text from variable layouts. Light mode will miss letterhead blocks and multi-column sections.
2. **Monetary amounts must include currency code**: Train the extractor on examples showing amounts with symbols (`$2,345.67`) or iso codes (`USD 2345.67`). Add examples to your extraction testing set showing what "correct" looks like in your region's format.
3. **Dates vary widely; normalize in post-processing**: Pension letters use different date formats (MM/DD/YYYY, DD-MMM-YYYY, spelled-out months). The extractor will return raw text; **always run output through a date parser** (e.g., `date-fns`, `chrono-node`) to normalize to ISO format before storing.
4. **Signatory name + title is one field**: Pension letters often print the signer's name and title together (e.g., "Sarah Johnson, VP of Pension Benefits"). Keep them in one field and parse programmatically in post-processing if you need to split.
5. **Enable review agent for every production run**: Set `advancedOptions.reviewAgent.enabled = true`. Pension data is legally binding; the review agent flags low-confidence extractions (e.g., "Could not locate pension amount" or "Multiple dates found"). Use these flags to route uncertain letters to human review before filing.
6. **Multimodal catches layout-based errors**: Enable `advancedMultimodalEnabled: true`. Some pension letters use visual boxes or tables to highlight the benefit amount. Multimodal extraction references both text and visual structure, reducing mistakes from OCR misalignment.
7. **import { ExtendClient } from "extend-ai";
import { z } from "zod";
import fs from "fs";
const client = new ExtendClient({ token: process.env.EXTEND_API_KEY });
// Zod schema for pension award letter extraction
const pensionAwardSchema = z.object({
recipient_name: z
.string()
.nullable()
.describe(
"Full name of the pension award recipient as printed on the letter"
),
recipient_address: z
.string()
.nullable()
.describe("Complete mailing address of the recipient including postal code"),
letter_date: z
.string()
.nullable()
.describe("Date the pension award letter was issued (ISO format or natural date)"),
pension_plan_name: z
.string()
.nullable()
.describe(
"Name of the pension plan under which the award is granted (e.g., 'Company Pension Plan 2024')"
),
employer_name: z
.string()
.nullable()
.describe(
"Name of the employer or organization issuing the pension (full legal name)"
),
years_of_service: z
.string()
.nullable()
.describe("Total years of service with the employer (numeric or text, e.g., '25 years')"),
monthly_pension_amount: z
.string()
.nullable()
.describe(
"Monthly pension payment amount in currency format (e.g., '$2,345.67' or '€1,500.00')"
),
commencement_date: z
.string()
.nullable()
.describe(
"Date when pension payments commence or have commenced (ISO format or natural date)"
),
payment_frequency: z
.string()
.nullable()
.describe(
"Frequency of pension payments (e.g., 'Monthly', 'Quarterly', 'Annually')"
),
issuing_department: z
.string()
.nullable()
.describe(
"Department or office issuing the pension award letter (e.g., 'Pension Administration Department')"
),
contact_email: z
.string()
.nullable()
.describe(
"Email address for inquiries or assistance regarding the pension award"
),
signatory_name: z
.string()
.nullable()
.describe(
"Name and title of the person signing the letter (e.g., 'Jane Smith, Director of Benefits')"
),
});
export async function processPensionAwardLetter(filePath: string) {
console.log(`Processing pension award letter: ${filePath}`);
// Convert local file to base64 data URL for SDK
const fileBuffer = fs.readFileSync(filePath);
const base64 = fileBuffer.toString("base64");
const dataUrl = `data:application/octet-stream;base64,${base64}`;
try {
// Step 1: Parse the pension award letter using agentic OCR
console.log("Step 1: Parsing pension award letter...");
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}`);
}
const markdown = parseRun.output.chunks
.map((chunk) => chunk.content)
.join("\n\n");
console.log("Parsed markdown (first 500 chars):");
console.log(markdown.substring(0, 500));
console.log("\n");
// Step 2: Extract structured fields from the pension award letter
console.log("Step 2: Extracting structured pension award fields...");
const extractRun = await client.extractRuns.createAndPoll({
file: { url: dataUrl },
config: {
schema: pensionAwardSchema,
baseProcessor: "extraction_performance",
advancedOptions: {
reviewAgent: {
enabled: true,
},
advancedMultimodalEnabled: true,
},
},
});
if (extractRun.status !== "PROCESSED") {
throw new Error(`Extraction failed with status: ${extractRun.status}`);
}
const extractedData = extractRun.output.value;
console.log("Extracted pension award data:");
console.log(JSON.stringify(extractedData, null, 2));
return {
success: true,
parseStatus: parseRun.status,
markdownPreview: markdown.substring(0, 500),
extractedFields: extractedData,
confidence: extractRun.confidence || "N/A",
};
} catch (error) {
console.error("Error processing pension award letter:", error);
throw error;
}
}
// Run the pipeline if this file is executed directly
const filePath = process.argv[2] || "./pension_award_letter.pdf";
processPensionAwardLetter(filePath).then((result) => {
console.log("\nPipeline completed successfully.");
console.log(JSON.stringify(result, null, 2));
});import os
import base64
import json
import sys
from extend_ai import Extend
client = Extend(token=os.environ["EXTEND_API_KEY"])
# Schema for pension award letter extraction
pension_award_schema = {
"type": "object",
"properties": {
"recipient_name": {
"type": ["string", "null"],
"description": "Full name of the pension award recipient as printed on the letter",
},
"recipient_address": {
"type": ["string", "null"],
"description": "Complete mailing address of the recipient including postal code",
},
"letter_date": {
"type": ["string", "null"],
"description": "Date the pension award letter was issued (ISO format or natural date)",
},
"pension_plan_name": {
"type": ["string", "null"],
"description": "Name of the pension plan under which the award is granted (e.g., 'Company Pension Plan 2024')",
},
"employer_name": {
"type": ["string", "null"],
"description": "Name of the employer or organization issuing the pension (full legal name)",
},
"years_of_service": {
"type": ["string", "null"],
"description": "Total years of service with the employer (numeric or text, e.g., '25 years')",
},
"monthly_pension_amount": {
"type": ["string", "null"],
"description": "Monthly pension payment amount in currency format (e.g., '$2,345.67' or '€1,500.00')",
},
"commencement_date": {
"type": ["string", "null"],
"description": "Date when pension payments commence or have commenced (ISO format or natural date)",
},
"payment_frequency": {
"type": ["string", "null"],
"description": "Frequency of pension payments (e.g., 'Monthly', 'Quarterly', 'Annually')",
},
"issuing_department": {
"type": ["string", "null"],
"description": "Department or office issuing the pension award letter (e.g., 'Pension Administration Department')",
},
"contact_email": {
"type": ["string", "null"],
"description": "Email address for inquiries or assistance regarding the pension award",
},
"signatory_name": {
"type": ["string", "null"],
"description": "Name and title of the person signing the letter (e.g., 'Jane Smith, Director of Benefits')",
},
},
}
def process_pension_award_letter(file_path: str) -> dict:
"""Process a pension award letter through parse and extract pipeline."""
print(f"Processing pension award letter: {file_path}")
# Convert local file to base64 data URL for SDK
with open(file_path, "rb") as f:
file_buffer = f.read()
base64_str = base64.b64encode(file_buffer).decode("utf-8")
data_url = f"data:application/octet-stream;base64,{base64_str}"
try:
# Step 1: Parse the pension award letter using agentic OCR
print("Step 1: Parsing pension award letter...")
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}")
markdown = "\n\n".join(chunk.content for chunk in parse_run.output.chunks)
print("Parsed markdown (first 500 chars):")
print(markdown[:500])
print()
# Step 2: Extract structured fields from the pension award letter
print("Step 2: Extracting structured pension award fields...")
extract_run = client.extract_runs.create_and_poll(
file={"url": data_url},
config={
"schema": pension_award_schema,
"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}")
extracted_data = extract_run.output.value
print("Extracted pension award data:")
print(json.dumps(extracted_data, indent=2))
return {
"success": True,
"parse_status": parse_run.status,
"markdown_preview": markdown[:500],
"extracted_fields": extracted_data,
"confidence": getattr(extract_run, "confidence", "N/A"),
}
except Exception as error:
print(f"Error processing pension award letter: {error}")
raise
if __name__ == "__main__":
file_path = sys.argv[1] if len(sys.argv) > 1 else "./pension_award_letter.pdf"
result = process_pension_award_letter(file_path)
print("\nPipeline completed successfully.")
print(json.dumps(result, indent=2))// This code uses Extend's REST API directly because Extend has no official Java SDK yet.
// It calls https://api.extend.ai endpoints with Bearer token authentication.
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.HashMap;
import java.util.Map;
public class PensionAwardLetterProcessor {
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();
public static void main(String[] args) throws Exception {
String filePath = args.length > 0 ? args[0] : "./pension_award_letter.pdf";
Map<String, Object> result = processPensionAwardLetter(filePath);
System.out.println("\nPipeline completed successfully.");
System.out.println(jsonPrettyPrint(result));
}
public static Map<String, Object> processPensionAwardLetter(String filePath)
throws Exception {
System.out.println("Processing pension award letter: " + filePath);
// Convert local file to base64 data URL
byte[] fileBytes = Files.readAllBytes(Paths.get(filePath));
String base64 = Base64.getEncoder().encodeToString(fileBytes);
String dataUrl = "data:application/octet-stream;base64," + base64;
try {
// Step 1: Parse the pension award letter using agentic OCR
System.out.println("Step 1: Parsing pension award letter...");
Map<String, Object> parseResponse = callParseRuns(dataUrl);
String parseStatus = (String) parseResponse.get("status");
if (!"PROCESSED".equals(parseStatus)) {
throw new Exception("Parse failed with status: " + parseStatus);
}
String markdown = extractMarkdownFromParseOutput(parseResponse);
System.out.println("Parsed markdown (first 500 chars):");
System.out.println(markdown.substring(0, Math.min(500, markdown.length())));
System.out.println();
// Step 2: Extract structured fields from the pension award letter
System.out.println("Step 2: Extracting structured pension award fields...");
Map<String, Object> extractResponse = callExtractRuns(dataUrl);
String extractStatus = (String) extractResponse.get("status");
if (!"PROCESSED".equals(extractStatus)) {
throw new Exception("Extraction failed with status: " + extractStatus);
}
Map<String, Object> extractedData =
(Map<String, Object>) ((Map<String, Object>) extractResponse.get("output")).get("value");
System.out.println("Extracted pension award data:");
System.out.println(jsonPrettyPrint(extractedData));
Map<String, Object> result = new HashMap<>();
result.put("success", true);
result.put("parseStatus", parseStatus);
result.put("markdownPreview", markdown.substring(0, Math.min(500, markdown.length())));
result.put("extractedFields", extractedData);
result.put("confidence", extractResponse.getOrDefault("confidence", "N/A"));
return result;
} catch (Exception error) {
System.err.println("Error processing pension award letter: " + error.getMessage());
throw error;
}
}
private static Map<String, Object> callParseRuns(String dataUrl) throws Exception {
String requestBody =
"{"
+ "\"file\":{\"url\":\""
+ dataUrl
+ "\"},"
+ "\"config\":{"
+ "\"blockOptions\":{\"text\":{\"agentic\":{\"enabled\":true}}},"
+ "\"chunkingStrategy\":{\"type\":\"document\"}"
+ "}"
+ "}";
HttpRequest request =
HttpRequest.newBuilder()
.uri(URI.create(API_BASE_URL + "/v1/parseRuns/createAndPoll"))
.header("Authorization", "Bearer " + API_KEY)
.header("Content-Type", "application/json")
.POST(HttpRequest.BodyPublishers.ofString(requestBody))
.build();
HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
return parseJsonResponse(response.body());
}
private static Map<String, Object> callExtractRuns(String dataUrl) throws Exception {
String schemaJson =
"{"
+ "\"type\":\"object\","
+ "\"properties\":{"
+ "\"recipient_name\":{\"type\":[\"string\",\"null\"],\"description\":\"Full name of the pension award recipient as printed on the letter\"},"
+ "\"recipient_address\":{\"type\":[\"string\",\"null\"],\"description\":\"Complete mailing address of the recipient including postal code\"},"
+ "\"letter_date\":{\"type\":[\"string\",\"null\"],\"description\":\"Date the pension award letter was issued (ISO format or natural date)\"},"
+ "\"pension_plan_name\":{\"type\":[\"string\",\"null\"],\"description\":\"Name of the pension plan under which the award is granted\"},"
+ "\"employer_name\":{\"type\":[\"string\",\"null\"],\"description\":\"Name of the employer or organization issuing the pension\"},"
+ "\"years_of_service\":{\"type\":[\"string\",\"null\"],\"description\":\"Total years of service with the employer\"},"
+ "\"monthly_pension_amount\":{\"type\":[\"string\",\"null\"],\"description\":\"Monthly pension payment amount in currency format\"},"
+ "\"commencement_date\":{\"type\":[\"string\",\"null\"],\"description\":\"Date when pension payments commence or have commenced\"},"
+ "\"payment_frequency\":{\"type\":[\"string\",\"null\"],\"description\":\"Frequency of pension payments\"},"
+ "\"issuing_department\":{\"type\":[\"string\",\"null\"],\"description\":\"Department or office issuing the pension award letter\"},"
+ "\"contact_email\":{\"type\":[\"string\",\"null\"],\"description\":\"Email address for inquiries or assistance regarding the pension award\"},"
+ "\"signatory_name\":{\"type\":[\"string\",\"null\"],\"description\":\"Name and title of the person signing the letter\"}"
+ "}"
+ "}";
String requestBody =
"{"
+ "\"file\":{\"url\":\""
+ dataUrl
+ "\"},"
+ "\"config\":{"
+ "\"schema\":"
+ schemaJson
+ ","
+ "\"baseProcessor\":\"extraction_performance\","
+ "\"advancedOptions\":{"
+ "\"reviewAgent\":{\"enabled\":true},"
+ "\"advancedMultimodalEnabled\":true"
+ "}"
+ "}"
+ "}";
HttpRequest request =
HttpRequest.newBuilder()
.uri(URI.create(API_BASE_URL + "/v1/extractRuns/createAndPoll"))
.header("Authorization", "Bearer " + API_KEY)
.header("Content-Type", "application/json")
.POST(HttpRequest.BodyPublishers.ofString(requestBody))
.build();
HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
return parseJsonResponse(response.body());
}
private static String extractMarkdownFromParseOutput(Map<String, Object> parseResponse) {
Map<String, Object> output = (Map<String, Object>) parseResponse.get("output");
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");
}
return markdown.toString();
}
private static Map<String, Object> parseJsonResponse(String jsonString) {
// Simple JSON parser for response objects
Map<String, Object> result = new HashMap<>();
jsonString = jsonString.trim();
if (jsonString.startsWith("{") && jsonString.endsWith("}")) {
jsonString = jsonString.substring(1, jsonString.length() - 1);
String[] pairs = jsonString.split(",(?=(?:[^\"]*\"[^\"]*\")*[^\"]*$)");
for (String pair : pairs) {
String[] keyValue = pair.split(":", 2);
if (keyValue.length == 2) {
String key = keyValue[0].trim().replaceAll("\"", "");
String value = keyValue[1].trim();
if (value.startsWith("\"") && value.endsWith("\"")) {
result.put(key, value.substring(1, value.length() - 1));
} else if ("true".equals(value) || "false".equals(value)) {
result.put(key, Boolean.parseBoolean(value));
} else if (value.startsWith("{") || value.startsWith("[")) {
result.put(key, value);
} else {
result.put(key, value);
}
}
}
}
return result;
}
private static String jsonPrettyPrint(Object obj) {
return obj.toString();
}
}// This code uses the Extend REST API directly because Extend has no official Go SDK yet.
// It calls https://api.extend.ai endpoints with standard net/http and encoding/json.
package main
import (
"bytes"
"encoding/base64"
"encoding/json"
"fmt"
"io"
"net/http"
"os"
"time"
)
const extendAPIBase = "https://api.extend.ai"
type ExtendClient struct {
token string
}
func NewExtendClient(token string) *ExtendClient {
return &ExtendClient{token: token}
}
func (c *ExtendClient) do(method, path string, body interface{}) ([]byte, error) {
url := extendAPIBase + path
var reqBody io.Reader
if body != nil {
jsonBody, err := json.Marshal(body)
if err != nil {
return nil, err
}
reqBody = bytes.NewReader(jsonBody)
}
req, err := http.NewRequest(method, url, reqBody)
if err != nil {
return nil, err
}
req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", c.token))
req.Header.Set("Content-Type", "application/json")
resp, err := http.DefaultClient.Do(req)
if err != nil {
return nil, err
}
defer resp.Body.Close()
respBody, err := io.ReadAll(resp.Body)
if err != nil {
return nil, err
}
if resp.StatusCode >= 400 {
return nil, fmt.Errorf("API error: %d %s", resp.StatusCode, string(respBody))
}
return respBody, nil
}
type ParseRunRequest struct {
File FileInput `json:"file"`
Config ParseConfig `json:"config"`
}
type FileInput struct {
URL string `json:"url"`
}
type ParseConfig struct {
BlockOptions BlockOptions `json:"blockOptions"`
ChunkingStrategy ChunkingStrategy `json:"chunkingStrategy"`
}
type BlockOptions struct {
Text TextOptions `json:"text"`
}
type TextOptions struct {
Agentic AgenticOptions `json:"agentic"`
}
type AgenticOptions struct {
Enabled bool `json:"enabled"`
}
type ChunkingStrategy struct {
Type string `json:"type"`
}
type ParseRunResponse struct {
Status string `json:"status"`
Output ParseOutput `json:"output"`
}
type ParseOutput struct {
Chunks []Chunk `json:"chunks"`
}
type Chunk struct {
Content string `json:"content"`
}
type ExtractRunRequest struct {
File FileInput `json:"file"`
Config ExtractConfig `json:"config"`
}
type ExtractConfig struct {
Schema map[string]interface{} `json:"schema"`
BaseProcessor string `json:"baseProcessor"`
AdvancedOptions AdvancedOptions `json:"advancedOptions"`
}
type AdvancedOptions struct {
ReviewAgent ReviewAgent `json:"reviewAgent"`
AdvancedMultimodalEnabled bool `json:"advancedMultimodalEnabled"`
}
type ReviewAgent struct {
Enabled bool `json:"enabled"`
}
type ExtractRunResponse struct {
Status string `json:"status"`
Output ExtractOutput `json:"output"`
Confidence interface{} `json:"confidence"`
}
type ExtractOutput struct {
Value map[string]interface{} `json:"value"`
}
type PipelineResult struct {
Success bool `json:"success"`
ParseStatus string `json:"parseStatus"`
MarkdownPreview string `json:"markdownPreview"`
ExtractedFields map[string]interface{} `json:"extractedFields"`
Confidence interface{} `json:"confidence"`
}
func ProcessPensionAwardLetter(filePath string) (*PipelineResult, error) {
fmt.Printf("Processing pension award letter: %s\n", filePath)
fileBuffer, err := os.ReadFile(filePath)
if err != nil {
return nil, err
}
base64Str := base64.StdEncoding.EncodeToString(fileBuffer)
dataURL := fmt.Sprintf("data:application/octet-stream;base64,%s", base64Str)
client := NewExtendClient(os.Getenv("EXTEND_API_KEY"))
// Step 1: Parse the pension award letter
fmt.Println("Step 1: Parsing pension award letter...")
parseReq := ParseRunRequest{
File: FileInput{URL: dataURL},
Config: ParseConfig{
BlockOptions: BlockOptions{
Text: TextOptions{
Agentic: AgenticOptions{Enabled: true},
},
},
ChunkingStrategy: ChunkingStrategy{Type: "document"},
},
}
parseBody, err := client.do("POST", "/v1/parseRuns", parseReq)
if err != nil {
return nil, err
}
var parseRun ParseRunResponse
if err := json.Unmarshal(parseBody, &parseRun); err != nil {
return nil, err
}
// Poll for parse completion
for parseRun.Status != "PROCESSED" && parseRun.Status != "FAILED" {
time.Sleep(2 * time.Second)
parseBody, err := client.do("GET", "/v1/parseRuns", nil)
if err != nil {
return nil, err
}
if err := json.Unmarshal(parseBody, &parseRun); err != nil {
return nil, err
}
}
if parseRun.Status != "PROCESSED" {
return nil, fmt.Errorf("parse failed with status: %s", parseRun.Status)
}
markdown := ""
for _, chunk := range parseRun.Output.Chunks {
markdown += chunk.Content + "\n\n"
}
fmt.Println("Parsed markdown (first 500 chars):")
if len(markdown) > 500 {
fmt.Println(markdown[:500])
} else {
fmt.Println(markdown)
}
fmt.Println()
// Step 2: Extract structured fields
fmt.Println("Step 2: Extracting structured pension award fields...")
schema := map[string]interface{}{
"type": "object",
"properties": map[string]interface{}{
"recipient_name": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Full name of the pension award recipient as printed on the letter",
},
"recipient_address": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Complete mailing address of the recipient including postal code",
},
"letter_date": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Date the pension award letter was issued (ISO format or natural date)",
},
"pension_plan_name": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Name of the pension plan under which the award is granted",
},
"employer_name": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Name of the employer or organization issuing the pension",
},
"years_of_service": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Total years of service with the employer",
},
"monthly_pension_amount": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Monthly pension payment amount in currency format",
},
"commencement_date": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Date when pension payments commence or have commenced",
},
"payment_frequency": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Frequency of pension payments",
},
"issuing_department": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Department or office issuing the pension award letter",
},
"contact_email": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Email address for inquiries or assistance regarding the pension award",
},
"signatory_name": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Name and title of the person signing the letter",
},
},
}
extractReq := ExtractRunRequest{
File: FileInput{URL: dataURL},
Config: ExtractConfig{
Schema: schema,
BaseProcessor: "extraction_performance",
AdvancedOptions: AdvancedOptions{
ReviewAgent: ReviewAgent{Enabled: true},
AdvancedMultimodalEnabled: true,
},
},
}
extractBody, err := client.do("POST", "/v1/extractRuns", extractReq)
if err != nil {
return nil, err
}
var extractRun ExtractRunResponse
if err := json.Unmarshal(extractBody, &extractRun); err != nil {
return nil, err
}
// Poll for extract completion
for extractRun.Status != "PROCESSED" && extractRun.Status != "FAILED" {
time.Sleep(2 * time.Second)
extractBody, err := client.do("GET", "/v1/extractRuns", nil)
if err != nil {
return nil, err
}
if err := json.Unmarshal(extractBody, &extractRun); err != nil {
return nil, err
}
}
if extractRun.Status != "PROCESSED" {
return nil, fmt.Errorf("extraction failed with status: %s", extractRun.Status)
}
fmt.Println("Extracted pension award data:")
extractedJSON, _ := json.MarshalIndent(extractRun.Output.Value, "", " ")
fmt.Println(string(extractedJSON))
markdownPreview := markdown
if len(markdown) > 500 {
markdownPreview = markdown[:500]
}
return &PipelineResult{
Success: true,
ParseStatus: parseRun.Status,
MarkdownPreview: markdownPreview,
ExtractedFields: extractRun.Output.Value,
Confidence: extractRun.Confidence,
}, nil
}
func main() {
filePath := "./pension_award_letter.pdf"
if len(os.Args) > 1 {
filePath = os.Args[1]
}
result, err := ProcessPensionAwardLetter(filePath)
if err != nil {
fmt.Fprintf(os.Stderr, "Error processing pension award letter: %v\n", err)
os.Exit(1)
}
fmt.Println("\nPipeline completed successfully.")
resultJSON, _ := json.MarshalIndent(result, "", " ")
fmt.Println(string(resultJSON))
}// Deploy the "Pension Award Letter" 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/pension-award-letter-extraction.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: pension-award-letter-extraction).
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, "pension-award-letter-extraction.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": "Pension Award Letter 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": {
"letter_date": {
"type": [
"string",
"null"
],
"description": "Date the pension award letter was issued"
},
"contact_email": {
"type": [
"string",
"null"
],
"description": "Email address for inquiries or assistance"
},
"employer_name": {
"type": [
"string",
"null"
],
"description": "Name of the employer/organization issuing the pension"
},
"recipient_name": {
"type": [
"string",
"null"
],
"description": "Full name of the pension award recipient"
},
"signatory_name": {
"type": [
"string",
"null"
],
"description": "Name and title of the person signing the letter"
},
"years_of_service": {
"type": [
"string",
"null"
],
"description": "Total years of service with the employer"
},
"commencement_date": {
"type": [
"string",
"null"
],
"description": "Date when pension payments commence"
},
"payment_frequency": {
"type": [
"string",
"null"
],
"description": "Frequency of pension payments (e.g., Monthly, Quarterly)"
},
"pension_plan_name": {
"type": [
"string",
"null"
],
"description": "Name of the pension plan under which the award is granted"
},
"recipient_address": {
"type": [
"string",
"null"
],
"description": "Complete mailing address of the recipient"
},
"issuing_department": {
"type": [
"string",
"null"
],
"description": "Department or office issuing the pension award letter"
},
"monthly_pension_amount": {
"type": [
"string",
"null"
],
"description": "Monthly pension payment amount in currency"
}
}
},
"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 extend_ai import Extend
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 / "pension-award-letter-extraction.json"
state: dict = {}
if STATE_FILE.exists():
with open(STATE_FILE, "r") as f:
state = json.load(f)
def save_state():
STATE_DIR.mkdir(parents=True, exist_ok=True)
with open(STATE_FILE, "w") as f:
json.dump(state, f, indent=2)
WORKFLOW = {
"name": "Pension Award Letter 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": {
"letter_date": {
"type": ["string", "null"],
"description": "Date the pension award letter was issued"
},
"contact_email": {
"type": ["string", "null"],
"description": "Email address for inquiries or assistance"
},
"employer_name": {
"type": ["string", "null"],
"description": "Name of the employer/organization issuing the pension"
},
"recipient_name": {
"type": ["string", "null"],
"description": "Full name of the pension award recipient"
},
"signatory_name": {
"type": ["string", "null"],
"description": "Name and title of the person signing the letter"
},
"years_of_service": {
"type": ["string", "null"],
"description": "Total years of service with the employer"
},
"commencement_date": {
"type": ["string", "null"],
"description": "Date when pension payments commence"
},
"payment_frequency": {
"type": ["string", "null"],
"description": "Frequency of pension payments (e.g., Monthly, Quarterly)"
},
"pension_plan_name": {
"type": ["string", "null"],
"description": "Name of the pension plan under which the award is granted"
},
"recipient_address": {
"type": ["string", "null"],
"description": "Complete mailing address of the recipient"
},
"issuing_department": {
"type": ["string", "null"],
"description": "Department or office issuing the pension award letter"
},
"monthly_pension_amount": {
"type": ["string", "null"],
"description": "Monthly pension payment amount in currency"
}
}
},
"baseProcessor": "extraction_performance",
"advancedOptions": {
"reviewAgent": {
"enabled": True
},
"advancedMultimodalEnabled": True
}
}
}
}
]
}
def main():
client = Extend(token=API_KEY)
print(f"Deploying \"{WORKFLOW['name']}\"…")
if state.get("workflowId"):
workflow_id = state["workflowId"]
print(f"✓ workflow already provisioned ({workflow_id}) — updating steps")
client.workflows.update(id=workflow_id, steps=WORKFLOW["steps"])
else:
# Try to find existing workflow with same name
try:
workflows_list = client.workflows.list(name=WORKFLOW["name"])
items = workflows_list.data if hasattr(workflows_list, "data") else (workflows_list.items if hasattr(workflows_list, "items") else [])
existing = next((x for x in items if getattr(x, "name", None) == WORKFLOW["name"]), None)
if existing and getattr(existing, "id", None):
state["workflowId"] = existing.id
save_state()
print(f"✓ workflow \"{WORKFLOW['name']}\" found in your account ({existing.id}) — updating steps")
client.workflows.update(id=existing.id, steps=WORKFLOW["steps"])
except Exception:
pass
if not state.get("workflowId"):
created = client.workflows.create(**WORKFLOW)
workflow_id = getattr(created, "id", None) or (getattr(created.workflow, "id", None) 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()
print(f"+ created workflow ({workflow_id})")
# Deploy the current draft as a new version
try:
client.workflows.create_version(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 code uses Extend's REST API directly because Extend has no official Java SDK yet.
// It calls https://api.extend.ai endpoints with Bearer token authentication.
import java.io.IOException;
import java.net.URI;
import java.net.URLEncoder;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
public class PensionAwardLetterProvisioning {
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("pension-award-letter-extraction.json");
private static final HttpClient HTTP_CLIENT = HttpClient.newHttpClient();
static class State {
String workflowId;
}
private static State state = new State();
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 && !state.workflowId.isEmpty()) {
System.out.println("✓ workflow already provisioned (" + state.workflowId + ") — updating steps");
Map<String, Object> updateBody = new LinkedHashMap<>();
updateBody.put("steps", workflow.get("steps"));
api("POST", "/workflows/" + state.workflowId, updateBody);
} else {
try {
String encodedName = URLEncoder.encode(workflowName, StandardCharsets.UTF_8);
Map<String, Object> listResponse = api("GET", "/workflows?name=" + encodedName, null);
List<?> items = (List<?>) listResponse.getOrDefault("data",
listResponse.getOrDefault("items", List.of()));
for (Object item : items) {
if (item instanceof Map) {
Map<?, ?> itemMap = (Map<?, ?>) item;
if (workflowName.equals(itemMap.get("name"))) {
String existingId = (String) itemMap.get("id");
if (existingId != null) {
state.workflowId = existingId;
saveState();
System.out.println("✓ workflow \"" + workflowName + "\" found in your account (" + existingId + ") — updating steps");
Map<String, Object> updateBody = new LinkedHashMap<>();
updateBody.put("steps", workflow.get("steps"));
api("POST", "/workflows/" + existingId, updateBody);
break;
}
}
}
}
} 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<?, ?> workflowObj = (Map<?, ?>) created.get("workflow");
if (workflowObj != null) {
wfId = (String) workflowObj.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 LinkedHashMap<>());
} 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.");
} catch (Exception e) {
System.err.println(e.getMessage() != null ? e.getMessage() : e.toString());
System.exit(1);
}
}
private static Map<String, Object> buildWorkflow() {
Map<String, Object> workflow = new LinkedHashMap<>();
workflow.put("name", "Pension Award Letter Processing Pipeline");
List<Map<String, Object>> steps = List.of(
buildTriggerStep(),
buildParseStep(),
buildExtractionStep()
);
workflow.put("steps", steps);
return workflow;
}
private static Map<String, Object> buildTriggerStep() {
Map<String, Object> step = new LinkedHashMap<>();
step.put("name", "startTrigger1");
step.put("type", "TRIGGER");
step.put("next", List.of(Map.of("step", "parse1")));
return step;
}
private static Map<String, Object> buildParseStep() {
Map<String, Object> step = new LinkedHashMap<>();
step.put("name", "parse1");
step.put("type", "PARSE");
Map<String, Object> parseConfig = new LinkedHashMap<>();
Map<String, Object> blockOptions = new LinkedHashMap<>();
Map<String, Object> textOptions = new LinkedHashMap<>();
Map<String, Object> agenticOptions = new LinkedHashMap<>();
agenticOptions.put("enabled", true);
textOptions.put("agentic", agenticOptions);
blockOptions.put("text", textOptions);
parseConfig.put("blockOptions", blockOptions);
Map<String, Object> chunkingStrategy = new LinkedHashMap<>();
chunkingStrategy.put("type", "document");
parseConfig.put("chunkingStrategy", chunkingStrategy);
Map<String, Object> config = new LinkedHashMap<>();
config.put("parseConfig", parseConfig);
step.put("config", config);
step.put("next", List.of(Map.of("step", "extraction2")));
return step;
}
private static Map<String, Object> buildExtractionStep() {
Map<String, Object> step = new LinkedHashMap<>();
step.put("name", "extraction2");
step.put("type", "EXTRACT");
Map<String, Object> schema = new LinkedHashMap<>();
schema.put("type", "object");
Map<String, Object> properties = new LinkedHashMap<>();
properties.put("letter_date", buildSchemaProperty("Date the pension award letter was issued"));
properties.put("contact_email", buildSchemaProperty("Email address for inquiries or assistance"));
properties.put("employer_name", buildSchemaProperty("Name of the employer/organization issuing the pension"));
properties.put("recipient_name", buildSchemaProperty("Full name of the pension award recipient"));
properties.put("signatory_name", buildSchemaProperty("Name and title of the person signing the letter"));
properties.put("years_of_service", buildSchemaProperty("Total years of service with the employer"));
properties.put("commencement_date", buildSchemaProperty("Date when pension payments commence"));
properties.put("payment_frequency", buildSchemaProperty("Frequency of pension payments (e.g., Monthly, Quarterly)"));
properties.put("pension_plan_name", buildSchemaProperty("Name of the pension plan under which the award is granted"));
properties.put("recipient_address", buildSchemaProperty("Complete mailing address of the recipient"));
properties.put("issuing_department", buildSchemaProperty("Department or office issuing the pension award letter"));
properties.put("monthly_pension_amount", buildSchemaProperty("Monthly pension payment amount in currency"));
schema.put("properties", properties);
Map<String, Object> extractorConfig = new LinkedHashMap<>();
extractorConfig.put("schema", schema);
extractorConfig.put("baseProcessor", "extraction_performance");
Map<String, Object> advancedOptions = new LinkedHashMap<>();
Map<String, Object> reviewAgent = new LinkedHashMap<>();
reviewAgent.put("enabled", true);
advancedOptions.put("reviewAgent", reviewAgent);
advancedOptions.put("advancedMultimodalEnabled", true);
extractorConfig.put("advancedOptions", advancedOptions);
Map<String, Object> config = new LinkedHashMap<>();
config.put("extractorConfig", extractorConfig);
step.put("config", config);
return step;
}
private static Map<String, Object> buildSchemaProperty(String description) {
Map<String, Object> prop = new LinkedHashMap<>();
prop.put("type", List.of("string", "null"));
prop.put("description", description);
return prop;
}
private static Map<String, Object> api(String method, String pathName, Object body)
throws IOException, InterruptedException {
String url = API + pathName;
HttpRequest.Builder requestBuilder = HttpRequest.newBuilder()
.uri(URI.create(url))
.method(method, body != null
? HttpRequest.BodyPublishers.ofString(toJson(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 = HTTP_CLIENT.send(request, HttpResponse.BodyHandlers.ofString());
Map<String, Object> data = parseJson(response.body());
if (response.statusCode() < 200 || response.statusCode() >= 300) {
String errorMsg = toJson(data);
if (errorMsg.length() > 300) {
errorMsg = errorMsg.substring(0, 300);
}
throw new RuntimeException(method + " " + pathName + " failed (" + response.statusCode() + "): " + errorMsg);
}
return data;
}
private static void loadState() throws IOException {
if (Files.exists(STATE_FILE)) {
String content = Files.readString(STATE_FILE);
Map<String, Object> parsed = parseJson(content);
state.workflowId = (String) parsed.get("workflowId");
}
}
private static void saveState() throws IOException {
Files.createDirectories(STATE_DIR);
Map<String, Object> stateMap = new LinkedHashMap<>();
if (state.workflowId != null) {
stateMap.put("workflowId", state.workflowId);
}
Files.writeString(STATE_FILE, toJson(stateMap));
}
private static String toJson(Object obj) {
if (obj == null) {
return "null";
}
if (obj instanceof String) {
return "\"" + escapeJson((String) obj) + "\"";
}
if (obj instanceof Number || obj instanceof Boolean) {
return obj.toString();
}
if (obj instanceof Map) {
Map<?, ?> map = (Map<?, ?>) obj;
StringBuilder sb = new StringBuilder("{");
boolean first = true;
for (Map.Entry<?, ?> entry : map.entrySet()) {
if (!first) sb.append(",");
sb.append("\"").append(escapeJson(entry.getKey().toString())).append("\":");
sb.append(toJson(entry.getValue()));
first = false;
}
sb.append("}");
return sb.toString();
}
if (obj instanceof List) {
List<?> list = (List<?>) obj;
StringBuilder sb = new StringBuilder("[");
boolean first = true;
for (Object item : list) {
if (!first) sb.append(",");
sb.append(toJson(item));
first = false;
}
sb.append("]");
return sb.toString();
}
return "null";
}
private static String escapeJson(String s) {
return s.replace("\\", "\\\\")
.replace("\"", "\\\"")
.replace("\n", "\\n")
.replace("\r", "\\r")
.replace("\t", "\\t");
}
@SuppressWarnings("unchecked")
private static Map<String, Object> parseJson(String json) {
json = json.trim();
if (!json.startsWith("{")) {
return new LinkedHashMap<>();
}
Map<String, Object> result = new LinkedHashMap<>();
int depth = 0;
int i = 1;
String currentKey = null;
StringBuilder currentValue = new StringBuilder();
boolean inString = false;
boolean escaped = false;
while (i < json.length() - 1) {
char c = json.charAt(i);
if (escaped) {
currentValue.append(c);
escaped = false;
i++;
continue;
}
if (c == '\\' && inString) {
escaped = true;
currentValue.append(c);
i++;
continue;
}
if (c == '"') {
inString = !inString;
currentValue.append(c);
i++;
continue;
}
if (inString) {
currentValue.append(c);
i++;
continue;
}
if (c == ':' && depth == 0 && currentKey == null) {
currentKey = currentValue.toString().trim();
if (currentKey.startsWith("\"") && currentKey.endsWith("\"")) {
currentKey = currentKey.substring(1, currentKey.length() - 1);
}
currentValue = new StringBuilder();
i++;
continue;
}
if ((c == ',' || c == '}') && depth == 0 && currentKey != null) {
String value = currentValue.toString().trim();
result.put(currentKey, parseJsonValue(value));
currentKey = null;
currentValue = new StringBuilder();
if (c == '}') break;
i++;
continue;
}
if (c == '{' || c == '[') depth++;
if (c == '}' || c == ']') depth--;
currentValue.append(c);
i++;
}
return result;
}
private static Object parseJsonValue(String value) {
value = value.trim();
if (value.isEmpty()) return null;
if ("null".equals(value)) return null;
if ("true".equals(value)) return true;
if ("false".equals(value)) return false;
if (value.startsWith("\"") && value.endsWith("\"")) {
return value.substring(1, value.length() - 1);
}
try {
if (value.contains(".")) {
return Double.parseDouble(value);
}
return Long.parseLong(value);
} catch (NumberFormatException e) {
return value;
}
}
}// This code uses the Extend REST API directly because Extend has no official Go SDK yet.
// It deploys the "Pension Award Letter" pipeline to your Extend account.
//
// Usage:
// export EXTEND_API_KEY=sk_... (from https://dashboard.extend.ai → API Keys)
// go run provision.go
//
// Generated by doc1 (template: pension-award-letter-extraction).
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
)
type State struct {
WorkflowID string `json:"workflowId,omitempty"`
}
var state State
func init() {
apiKey = os.Getenv("EXTEND_API_KEY")
if apiKey == "" {
fmt.Fprintf(os.Stderr, "Set EXTEND_API_KEY first.\n")
os.Exit(1)
}
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, "pension-award-letter-extraction.json")
// Load existing state if it exists
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 bodyReader io.Reader
var contentType string
if body != nil {
bodyBytes, err := json.Marshal(body)
if err != nil {
return nil, err
}
bodyReader = bytes.NewReader(bodyBytes)
contentType = "application/json"
}
req, err := http.NewRequest(method, API+pathName, bodyReader)
if err != nil {
return nil, err
}
req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", apiKey))
req.Header.Set("x-extend-api-version", VERSION)
if contentType != "" {
req.Header.Set("Content-Type", contentType)
}
client := &http.Client{}
resp, err := client.Do(req)
if err != nil {
return nil, err
}
defer resp.Body.Close()
respBody, err := io.ReadAll(resp.Body)
if err != nil {
return nil, err
}
var data map[string]interface{}
json.Unmarshal(respBody, &data)
if resp.StatusCode >= 400 {
respStr := string(respBody)
if len(respStr) > 300 {
respStr = respStr[:300]
}
return nil, fmt.Errorf("%s %s failed (%d): %s", method, pathName, resp.StatusCode, respStr)
}
return data, nil
}
func main() {
workflow := map[string]interface{}{
"name": "Pension Award Letter 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{}{
"letter_date": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Date the pension award letter was issued",
},
"contact_email": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Email address for inquiries or assistance",
},
"employer_name": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Name of the employer/organization issuing the pension",
},
"recipient_name": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Full name of the pension award recipient",
},
"signatory_name": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Name and title of the person signing the letter",
},
"years_of_service": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Total years of service with the employer",
},
"commencement_date": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Date when pension payments commence",
},
"payment_frequency": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Frequency of pension payments (e.g., Monthly, Quarterly)",
},
"pension_plan_name": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Name of the pension plan under which the award is granted",
},
"recipient_address": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Complete mailing address of the recipient",
},
"issuing_department": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Department or office issuing the pension award letter",
},
"monthly_pension_amount": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Monthly pension payment amount in currency",
},
},
},
"baseProcessor": "extraction_performance",
"advancedOptions": map[string]interface{}{
"reviewAgent": map[string]interface{}{
"enabled": true,
},
"advancedMultimodalEnabled": true,
},
},
},
},
},
}
workflowName := workflow["name"].(string)
fmt.Printf("Deploying \"%s\"…\n", workflowName)
if state.WorkflowID != "" {
fmt.Printf("✓ workflow already provisioned (%s) — updating steps\n", state.WorkflowID)
_, err := apiCall("POST", fmt.Sprintf("/workflows/%s", state.WorkflowID), map[string]interface{}{
"steps": workflow["steps"],
})
if err != nil {
fmt.Fprintf(os.Stderr, "%v\n", err)
os.Exit(1)
}
} else {
// Try to find an existing workflow with the same name
listPath := fmt.Sprintf("/workflows?name=%s", url.QueryEscape(workflowName))
listResp, err := apiCall("GET", listPath, nil)
if err == nil {
var items []map[string]interface{}
if data, ok := listResp["data"].([]interface{}); ok {
for _, item := range data {
items = append(items, item.(map[string]interface{}))
}
} else if data, ok := listResp["items"].([]interface{}); ok {
for _, item := range data {
items = append(items, item.(map[string]interface{}))
}
}
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)
_, err := apiCall("POST", fmt.Sprintf("/workflows/%s", id), map[string]interface{}{
"steps": workflow["steps"],
})
if err != nil {
fmt.Fprintf(os.Stderr, "%v\n", err)
os.Exit(1)
}
break
}
}
}
}
if state.WorkflowID == "" {
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 (best-effort)
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.")
}A Pension Award Letter is an official document issued by government or corporate pension administrators to inform employees of their approved pension benefits. This template captures the beneficiary's personal details, pension plan information, monthly award amounts, commencement dates, and payment terms.