Extracts sole proprietorship business income and expense information from IRS Schedule C.
Schedule C is a tax form filed by sole proprietors to report net profit or loss from self-employment business operations, including business identification, income sources, itemized expenses, and accounting method used. This template takes in Schedule C (Form 1040) and outputs markdown (.md) capturing the form's full text and layout, and JSON (.json) with structured business and financial fields including proprietor details, income, expenses, and net profit/loss 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": "Schedule C (Form 1040) 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": {
"ssn": {
"type": [
"string",
"null"
],
"description": "Social Security Number of the proprietor"
},
"gross_income": {
"type": [
"number",
"null"
],
"description": "Gross income from business operations"
},
"business_code": {
"type": [
"string",
"null"
],
"description": "Business code from IRS instructions"
},
"business_name": {
"type": [
"string",
"null"
],
"description": "Separate business name if applicable"
},
"total_expenses": {
"type": [
"number",
"null"
],
"description": "Total business expenses"
},
"net_profit_loss": {
"type": [
"number",
"null"
],
"description": "Net profit or loss from business"
},
"proprietor_name": {
"type": [
"string",
"null"
],
"description": "Name of the sole proprietor"
},
"business_address": {
"type": [
"string",
"null"
],
"description": "Business address including city, state, and ZIP code"
},
"accounting_method": {
"type": [
"string",
"null"
],
"description": "Accounting method used: Cash, Accrual, or Other"
},
"form_1099_required": {
"type": [
"boolean",
"null"
],
"description": "Whether Form 1099 filing is required"
},
"principal_business": {
"type": [
"string",
"null"
],
"description": "Principal business or profession, including product or service"
},
"material_participation": {
"type": [
"boolean",
"null"
],
"description": "Whether proprietor materially participated in business operation"
}
}
},
"baseProcessor": "extraction_performance",
"advancedOptions": {
"reviewAgent": {
"enabled": true
},
"advancedMultimodalEnabled": true
}
}
}
}
]
}# Schedule C (Form 1040) Processing — Extend AI Skill
## What this pipeline does
This pipeline extracts key business information and financial data from Schedule C (Form 1040) tax documents. It parses the form to markdown with agentic OCR handling (for handwritten entries and complex layouts), then extracts structured fields including proprietor details, business identification, income sources, total expenses, and net profit/loss into a JSON object ready for tax software integration or compliance review.
## When to use this
- **Tax preparation workflows**: Automated intake of Schedule C forms from individual tax clients before CPA review
- **Business loan applications**: Rapid extraction of financial summaries (gross income, net profit) for underwriting systems
- **Accounting software integration**: Bulk import of Schedule C data into bookkeeping platforms (QuickBooks, Xero, FreshBooks)
- **Tax compliance audits**: Programmatic validation of reported income, expenses, and business codes across batches of returns
- **IRS filing assistance**: Pre-population of tax software fields to reduce manual data entry errors
## Processor pipeline
### Step 1: Parse (agentic OCR mode)
- **Processor**: `parse_performance` with `agentic: true`
- **Purpose**: Convert Schedule C PDF or scanned image to clean markdown, preserving table structure and handling handwritten fields
- **Why this config**: Schedule C forms are commonly scanned or printed with handwriting (proprietor name, business address). Agentic OCR intelligently reconstructs section layout (Part I: Income, Part II: Expenses) without losing numerical accuracy. Document-level chunking keeps the full form as one logical unit for accurate field relationships.
- **Output**: Markdown with preserved tables and line items
### Step 2: Extract (performance mode with advanced options)
- **Processor**: `extraction_performance` with `reviewAgent: true` + `advancedMultimodalEnabled: true`
- **Purpose**: Pull 11 structured fields (proprietor name, SSN, business details, financials, accounting method, participation status) into a validated JSON schema
- **Why this config**: Tax forms require high accuracy for sensitive PII (SSN) and critical financial figures (gross income, net profit). The review agent catches inconsistencies (e.g., net profit ≠ gross income − expenses). Advanced multimodal processing reads both text and form box values, essential for pre-printed checkbox fields (accounting method, material participation).
- **Output**: Fully typed JSON object matching the schema below
## 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 });
// Define the Schedule C extraction schema using Zod
const scheduleCSchema = z.object({
proprietor_name: z.string().nullable().describe(
"Full legal name of the sole proprietor as it appears on the form"
),
ssn: z.string().nullable().describe(
"Social Security Number of the proprietor (9-digit format or redacted)"
),
principal_business: z.string().nullable().describe(
"Principal business or profession, including product or service description"
),
business_code: z.string().nullable().describe(
"6-digit IRS business code from Form 1040 Schedule C instructions"
),
business_name: z.string().nullable().describe(
"Separate business name (if different from proprietor name)"
),
business_address: z.string().nullable().describe(
"Complete business address including street, city, state, and ZIP code"
),
gross_income: z.number().nullable().describe(
"Gross income from business operations (numeric value in dollars, no currency symbol)"
),
total_expenses: z.number().nullable().describe(
"Total business expenses reported in Part II (numeric value in dollars)"
),
net_profit_loss: z.number().nullable().describe(
"Net profit or loss (bottom line: gross income minus total expenses, can be negative)"
),
accounting_method: z.string().nullable().describe(
"Accounting method used: 'Cash', 'Accrual', or 'Other' as indicated on the form"
),
material_participation: z.boolean().nullable().describe(
"True if proprietor materially participated in business operation (Box B, Part IV)"
),
form_1099_required: z.boolean().nullable().describe(
"True if Form 1099 filing is required for the business"
),
});
type ScheduleC = z.infer<typeof scheduleCSchema>;
/**
* Process a Schedule C (Form 1040) document through parse + extract pipeline
* @param filePath - Local file path to Schedule C PDF or image
* @returns Extracted Schedule C data
*/
export async function processScheduleCForm1040(filePath: string): Promise<ScheduleC> {
console.log(`Processing Schedule C form: ${filePath}`);
// Read local file and convert to data URL for upload
const fileBuffer = fs.readFileSync(filePath);
const base64Data = fileBuffer.toString("base64");
const dataUrl = `data:application/octet-stream;base64,${base64Data}`;
// Step 1: Parse the Schedule C form to markdown with agentic OCR
console.log("Step 1: Parsing Schedule C form...");
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 run failed with status: ${parseRun.status}`);
}
// Log parsed markdown for debugging
const parsedMarkdown = parseRun.output.chunks
.map((chunk) => chunk.content)
.join("\n\n");
console.log("Parsed markdown preview (first 500 chars):");
console.log(parsedMarkdown.substring(0, 500) + "...\n");
// Step 2: Extract structured fields using Zod schema with performance mode
console.log("Step 2: Extracting Schedule C fields...");
const extractRun = await client.extractRuns.createAndPoll({
file: { url: dataUrl },
config: {
schema: scheduleCSchema,
baseProcessor: "extraction_performance",
advancedOptions: {
reviewAgent: {
enabled: true,
},
advancedMultimodalEnabled: true,
},
},
});
if (extractRun.status !== "PROCESSED") {
throw new Error(`Extract run failed with status: ${extractRun.status}`);
}
const extractedData: ScheduleC = extractRun.output.value;
// Validation: Ensure financial calculations are consistent
if (
extractedData.gross_income !== null &&
extractedData.total_expenses !== null &&
extractedData.net_profit_loss !== null
) {
const calculatedNetProfit =
extractedData.gross_income - extractedData.total_expenses;
const difference = Math.abs(calculatedNetProfit - extractedData.net_profit_loss);
if (difference > 1) {
// Allow $1 rounding difference
console.warn(
`⚠️ Financial consistency check: Net profit mismatch detected.` +
` Calculated: ${calculatedNetProfit}, Extracted: ${extractedData.net_profit_loss}`
);
}
}
console.log("\n✅ Schedule C extraction complete:");
console.log(JSON.stringify(extractedData, null, 2));
return extractedData;
}
// Example usage (if running directly)
if (require.main === module) {
const testFilePath = process.argv[2] || "./schedule_c_sample.pdf";
processScheduleCForm1040(testFilePath)
.then((result) => {
console.log("\n📋 Final extracted Schedule C data:");
console.log(JSON.stringify(result, null, 2));
})
.catch((err) => {
console.error("❌ Error processing Schedule C:", err.message);
process.exit(1);
});
}
```
---
## CLI equivalent
```bash
# Set API key
export EXTEND_API_KEY="sk_..."
# Step 1: Parse Schedule C to markdown
extend parse schedule_c_2024.pdf > schedule_c_parsed.md
# Step 2: Extract structured fields using the schema file
extend extract schedule_c_2024.pdf --schema schedule_c_schema.json
```
### Schema file (schedule_c_schema.json)
```json
{
"type": "object",
"properties": {
"proprietor_name": {
"type": ["string", "null"],
"description": "Full legal name of the sole proprietor as it appears on the form"
},
"ssn": {
"type": ["string", "null"],
"description": "Social Security Number of the proprietor (9-digit format or redacted)"
},
"principal_business": {
"type": ["string", "null"],
"description": "Principal business or profession, including product or service description"
},
"business_code": {
"type": ["string", "null"],
"description": "6-digit IRS business code from Form 1040 Schedule C instructions"
},
"business_name": {
"type": ["string", "null"],
"description": "Separate business name (if different from proprietor name)"
},
"business_address": {
"type": ["string", "null"],
"description": "Complete business address including street, city, state, and ZIP code"
},
"gross_income": {
"type": ["number", "null"],
"description": "Gross income from business operations (numeric value in dollars, no currency symbol)"
},
"total_expenses": {
"type": ["number", "null"],
"description": "Total business expenses reported in Part II (numeric value in dollars)"
},
"net_profit_loss": {
"type": ["number", "null"],
"description": "Net profit or loss (bottom line: gross income minus total expenses, can be negative)"
},
"accounting_method": {
"type": ["string", "null"],
"description": "Accounting method used: 'Cash', 'Accrual', or 'Other' as indicated on the form"
},
"material_participation": {
"type": ["boolean", "null"],
"description": "True if proprietor materially participated in business operation (Box B, Part IV)"
},
"form_1099_required": {
"type": ["boolean", "null"],
"description":import { ExtendClient } from "extend-ai";
import { z } from "zod";
import fs from "fs";
const client = new ExtendClient({ token: process.env.EXTEND_API_KEY });
// Define the Schedule C extraction schema using Zod
const scheduleCSchema = z.object({
proprietor_name: z.string().nullable().describe(
"Full legal name of the sole proprietor as it appears on the form"
),
ssn: z.string().nullable().describe(
"Social Security Number of the proprietor (9-digit format or redacted)"
),
principal_business: z.string().nullable().describe(
"Principal business or profession, including product or service description"
),
business_code: z.string().nullable().describe(
"6-digit IRS business code from Form 1040 Schedule C instructions"
),
business_name: z.string().nullable().describe(
"Separate business name (if different from proprietor name)"
),
business_address: z.string().nullable().describe(
"Complete business address including street, city, state, and ZIP code"
),
gross_income: z.number().nullable().describe(
"Gross income from business operations (numeric value in dollars, no currency symbol)"
),
total_expenses: z.number().nullable().describe(
"Total business expenses reported in Part II (numeric value in dollars)"
),
net_profit_loss: z.number().nullable().describe(
"Net profit or loss (bottom line: gross income minus total expenses, can be negative)"
),
accounting_method: z.string().nullable().describe(
"Accounting method used: 'Cash', 'Accrual', or 'Other' as indicated on the form"
),
material_participation: z.boolean().nullable().describe(
"True if proprietor materially participated in business operation (Box B, Part IV)"
),
form_1099_required: z.boolean().nullable().describe(
"True if Form 1099 filing is required for the business"
),
});
type ScheduleC = z.infer<typeof scheduleCSchema>;
/**
* Process a Schedule C (Form 1040) document through parse + extract pipeline
* @param filePath - Local file path to Schedule C PDF or image
* @returns Extracted Schedule C data
*/
export async function processScheduleCForm1040(filePath: string): Promise<ScheduleC> {
console.log(`Processing Schedule C form: ${filePath}`);
// Read local file and convert to data URL for upload
const fileBuffer = fs.readFileSync(filePath);
const base64Data = fileBuffer.toString("base64");
const dataUrl = `data:application/octet-stream;base64,${base64Data}`;
// Step 1: Parse the Schedule C form to markdown with agentic OCR
console.log("Step 1: Parsing Schedule C form...");
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 run failed with status: ${parseRun.status}`);
}
// Log parsed markdown for debugging
const parsedMarkdown = parseRun.output.chunks
.map((chunk) => chunk.content)
.join("\n\n");
console.log("Parsed markdown preview (first 500 chars):");
console.log(parsedMarkdown.substring(0, 500) + "...\n");
// Step 2: Extract structured fields using Zod schema with performance mode
console.log("Step 2: Extracting Schedule C fields...");
const extractRun = await client.extractRuns.createAndPoll({
file: { url: dataUrl },
config: {
schema: scheduleCSchema,
baseProcessor: "extraction_performance",
advancedOptions: {
reviewAgent: {
enabled: true,
},
advancedMultimodalEnabled: true,
},
},
});
if (extractRun.status !== "PROCESSED") {
throw new Error(`Extract run failed with status: ${extractRun.status}`);
}
const extractedData: ScheduleC = extractRun.output.value;
// Validation: Ensure financial calculations are consistent
if (
extractedData.gross_income !== null &&
extractedData.total_expenses !== null &&
extractedData.net_profit_loss !== null
) {
const calculatedNetProfit =
extractedData.gross_income - extractedData.total_expenses;
const difference = Math.abs(calculatedNetProfit - extractedData.net_profit_loss);
if (difference > 1) {
// Allow $1 rounding difference
console.warn(
`⚠️ Financial consistency check: Net profit mismatch detected.` +
` Calculated: ${calculatedNetProfit}, Extracted: ${extractedData.net_profit_loss}`
);
}
}
console.log("\n✅ Schedule C extraction complete:");
console.log(JSON.stringify(extractedData, null, 2));
return extractedData;
}import os
import json
from typing import Optional
from extend_ai import Extend
client = Extend(token=os.environ["EXTEND_API_KEY"])
# Define the Schedule C extraction schema as a dictionary
schedule_c_schema = {
"type": "object",
"properties": {
"proprietor_name": {
"type": ["string", "null"],
"description": "Full legal name of the sole proprietor as it appears on the form",
},
"ssn": {
"type": ["string", "null"],
"description": "Social Security Number of the proprietor (9-digit format or redacted)",
},
"principal_business": {
"type": ["string", "null"],
"description": "Principal business or profession, including product or service description",
},
"business_code": {
"type": ["string", "null"],
"description": "6-digit IRS business code from Form 1040 Schedule C instructions",
},
"business_name": {
"type": ["string", "null"],
"description": "Separate business name (if different from proprietor name)",
},
"business_address": {
"type": ["string", "null"],
"description": "Complete business address including street, city, state, and ZIP code",
},
"gross_income": {
"type": ["number", "null"],
"description": "Gross income from business operations (numeric value in dollars, no currency symbol)",
},
"total_expenses": {
"type": ["number", "null"],
"description": "Total business expenses reported in Part II (numeric value in dollars)",
},
"net_profit_loss": {
"type": ["number", "null"],
"description": "Net profit or loss (bottom line: gross income minus total expenses, can be negative)",
},
"accounting_method": {
"type": ["string", "null"],
"description": "Accounting method used: 'Cash', 'Accrual', or 'Other' as indicated on the form",
},
"material_participation": {
"type": ["boolean", "null"],
"description": "True if proprietor materially participated in business operation (Box B, Part IV)",
},
"form_1099_required": {
"type": ["boolean", "null"],
"description": "True if Form 1099 filing is required for the business",
},
},
}
def process_schedule_c_form_1040(file_path: str) -> dict:
"""
Process a Schedule C (Form 1040) document through parse + extract pipeline
Args:
file_path: Local file path to Schedule C PDF or image
Returns:
Extracted Schedule C data as a dictionary
"""
print(f"Processing Schedule C form: {file_path}")
# Read local file and convert to data URL for upload
with open(file_path, "rb") as f:
file_buffer = f.read()
base64_data = __import__("base64").b64encode(file_buffer).decode("utf-8")
data_url = f"data:application/octet-stream;base64,{base64_data}"
# Step 1: Parse the Schedule C form to markdown with agentic OCR
print("Step 1: Parsing Schedule C form...")
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 run failed with status: {parse_run.status}")
# Log parsed markdown for debugging
parsed_markdown = "\n\n".join(chunk.content for chunk in parse_run.output.chunks)
print("Parsed markdown preview (first 500 chars):")
print(parsed_markdown[:500] + "...\n")
# Step 2: Extract structured fields using schema with performance mode
print("Step 2: Extracting Schedule C fields...")
extract_run = client.extract_runs.create_and_poll(
file={"url": data_url},
config={
"schema": schedule_c_schema,
"baseProcessor": "extraction_performance",
"advancedOptions": {
"reviewAgent": {
"enabled": True,
},
"advancedMultimodalEnabled": True,
},
},
)
if extract_run.status != "PROCESSED":
raise Exception(f"Extract run failed with status: {extract_run.status}")
extracted_data = extract_run.output.value
# Validation: Ensure financial calculations are consistent
if (
extracted_data.get("gross_income") is not None
and extracted_data.get("total_expenses") is not None
and extracted_data.get("net_profit_loss") is not None
):
calculated_net_profit = (
extracted_data["gross_income"] - extracted_data["total_expenses"]
)
difference = abs(calculated_net_profit - extracted_data["net_profit_loss"])
if difference > 1: # Allow $1 rounding difference
print(
f"⚠️ Financial consistency check: Net profit mismatch detected."
f" Calculated: {calculated_net_profit}, Extracted: {extracted_data['net_profit_loss']}"
)
print("\n✅ Schedule C extraction complete:")
print(json.dumps(extracted_data, indent=2))
return extracted_data// This code uses Extend's REST API directly because Extend has no official Java SDK yet.
// It calls https://api.extend.ai endpoints using only java.net.http.HttpClient (no 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.Map;
import java.util.List;
public class ScheduleCProcessor {
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 class ScheduleC {
public String proprietor_name;
public String ssn;
public String principal_business;
public String business_code;
public String business_name;
public String business_address;
public Double gross_income;
public Double total_expenses;
public Double net_profit_loss;
public String accounting_method;
public Boolean material_participation;
public Boolean form_1099_required;
@Override
public String toString() {
return "ScheduleC{" +
"proprietor_name='" + proprietor_name + '\'' +
", ssn='" + ssn + '\'' +
", principal_business='" + principal_business + '\'' +
", business_code='" + business_code + '\'' +
", business_name='" + business_name + '\'' +
", business_address='" + business_address + '\'' +
", gross_income=" + gross_income +
", total_expenses=" + total_expenses +
", net_profit_loss=" + net_profit_loss +
", accounting_method='" + accounting_method + '\'' +
", material_participation=" + material_participation +
", form_1099_required=" + form_1099_required +
'}';
}
}
/**
* Process a Schedule C (Form 1040) document through parse + extract pipeline
* @param filePath - Local file path to Schedule C PDF or image
* @return Extracted Schedule C data
*/
public static ScheduleC processScheduleCForm1040(String filePath) throws IOException, InterruptedException {
System.out.println("Processing Schedule C form: " + filePath);
// Read local file and convert to data URL for upload
byte[] fileBytes = Files.readAllBytes(Paths.get(filePath));
String base64Data = Base64.getEncoder().encodeToString(fileBytes);
String dataUrl = "data:application/octet-stream;base64," + base64Data;
// Step 1: Parse the Schedule C form to markdown with agentic OCR
System.out.println("Step 1: Parsing Schedule C form...");
String parseRequestBody = buildParseRequestBody(dataUrl);
String parseRunId = createAndPollParseRun(parseRequestBody);
// Retrieve parse run output
String parseOutput = getParseRunOutput(parseRunId);
String parsedMarkdown = extractMarkdownFromParseOutput(parseOutput);
System.out.println("Parsed markdown preview (first 500 chars):");
System.out.println(parsedMarkdown.substring(0, Math.min(500, parsedMarkdown.length())) + "...\n");
// Step 2: Extract structured fields using schema with performance mode
System.out.println("Step 2: Extracting Schedule C fields...");
String extractRequestBody = buildExtractRequestBody(dataUrl);
String extractRunId = createAndPollExtractRun(extractRequestBody);
// Retrieve extract run output
String extractOutput = getExtractRunOutput(extractRunId);
ScheduleC extractedData = parseExtractedData(extractOutput);
// Validation: Ensure financial calculations are consistent
if (extractedData.gross_income != null &&
extractedData.total_expenses != null &&
extractedData.net_profit_loss != null) {
double calculatedNetProfit = extractedData.gross_income - extractedData.total_expenses;
double difference = Math.abs(calculatedNetProfit - extractedData.net_profit_loss);
if (difference > 1) {
System.out.println("⚠️ Financial consistency check: Net profit mismatch detected." +
" Calculated: " + calculatedNetProfit + ", Extracted: " + extractedData.net_profit_loss);
}
}
System.out.println("\n✅ Schedule C extraction complete:");
System.out.println(extractedData.toString());
return extractedData;
}
private static String buildParseRequestBody(String dataUrl) {
return "{" +
"\"file\":{\"url\":\"" + escapeJson(dataUrl) + "\"}," +
"\"config\":{" +
"\"blockOptions\":{" +
"\"text\":{" +
"\"agentic\":{\"enabled\":true}" +
"}" +
"}," +
"\"chunkingStrategy\":{\"type\":\"document\"}" +
"}" +
"}";
}
private static String buildExtractRequestBody(String dataUrl) {
return "{" +
"\"file\":{\"url\":\"" + escapeJson(dataUrl) + "\"}," +
"\"config\":{" +
"\"schema\":{" +
"\"type\":\"object\"," +
"\"properties\":{" +
"\"proprietor_name\":{\"type\":[\"string\",\"null\"],\"description\":\"Name of the sole proprietor\"}," +
"\"ssn\":{\"type\":[\"string\",\"null\"],\"description\":\"Social Security Number of the proprietor\"}," +
"\"principal_business\":{\"type\":[\"string\",\"null\"],\"description\":\"Principal business or profession\"}," +
"\"business_code\":{\"type\":[\"string\",\"null\"],\"description\":\"Business code from IRS instructions\"}," +
"\"business_name\":{\"type\":[\"string\",\"null\"],\"description\":\"Separate business name if applicable\"}," +
"\"business_address\":{\"type\":[\"string\",\"null\"],\"description\":\"Business address\"}," +
"\"gross_income\":{\"type\":[\"number\",\"null\"],\"description\":\"Gross income from business operations\"}," +
"\"total_expenses\":{\"type\":[\"number\",\"null\"],\"description\":\"Total business expenses\"}," +
"\"net_profit_loss\":{\"type\":[\"number\",\"null\"],\"description\":\"Net profit or loss from business\"}," +
"\"accounting_method\":{\"type\":[\"string\",\"null\"],\"description\":\"Accounting method used\"}," +
"\"material_participation\":{\"type\":[\"boolean\",\"null\"],\"description\":\"Whether proprietor materially participated\"}," +
"\"form_1099_required\":{\"type\":[\"boolean\",\"null\"],\"description\":\"Whether Form 1099 filing is required\"}" +
"}" +
"}," +
"\"baseProcessor\":\"extraction_performance\"," +
"\"advancedOptions\":{" +
"\"reviewAgent\":{\"enabled\":true}," +
"\"advancedMultimodalEnabled\":true" +
"}" +
"}" +
"}";
}
private static String createAndPollParseRun(String requestBody) throws IOException, InterruptedException {
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());
if (response.statusCode() != 200) {
throw new RuntimeException("Parse run failed with status: " + response.statusCode() + " " + response.body());
}
return extractRunId(response.body());
}
private static String createAndPollExtractRun(String requestBody) throws IOException, InterruptedException {
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());
if (response.statusCode() != 200) {
throw new RuntimeException("Extract run failed with status: " + response.statusCode() + " " + response.body());
}
return extractRunId(response.body());
}
private static String getParseRunOutput(String runId) throws IOException, InterruptedException {
HttpRequest request = HttpRequest.newBuilder()
.uri(URI.create(API_BASE_URL + "/v1/parseRuns/" + runId))
.header("Authorization", "Bearer " + API_KEY)
.GET()
.build();
HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
if (response.statusCode() != 200) {
throw new RuntimeException("Failed to retrieve parse run: " + response.statusCode());
}
return response.body();
}
private static String getExtractRunOutput(String runId) throws IOException, InterruptedException {
HttpRequest request = HttpRequest.newBuilder()
.uri(URI.create(API_BASE_URL + "/v1/extractRuns/" + runId))
.header("Authorization", "Bearer " + API_KEY)
.GET()
.build();
HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
if (response.statusCode() != 200) {
throw new RuntimeException("Failed to retrieve extract run: " + response.statusCode());
}
return response.body();
}
private static String extractRunId(String jsonResponse) {
int idIndex = jsonResponse.indexOf("\"id\":\"");
if (idIndex == -1) {
throw new RuntimeException("Could not extract run ID from response");
}
int startIndex = idIndex + 6;
int endIndex = jsonResponse.indexOf("\"", startIndex);
return jsonResponse.substring(startIndex, endIndex);
}
private static String extractMarkdownFromParseOutput(String jsonResponse) {
int chunksIndex = jsonResponse.indexOf("\"chunks\":");
if (chunksIndex == -1) {
return "";
}
StringBuilder markdown = new StringBuilder();
int contentIndex = jsonResponse.indexOf("\"content\":\"", chunksIndex);
while (contentIndex != -1) {
int startIndex = contentIndex + 11;
int endIndex = jsonResponse.indexOf("\"", startIndex);
String content = jsonResponse.substring(startIndex, endIndex);
content = content.replace("\\n", "\n").replace("\\\"", "\"");
markdown.append(content).append("\n\n");
contentIndex = jsonResponse.indexOf("\"content\":\"", endIndex);
}
return markdown.toString();
}
private static ScheduleC parseExtractedData(String jsonResponse) {
ScheduleC result = new ScheduleC();
result.proprietor_name = extractStringField(jsonResponse, "proprietor_name");
result.ssn = extractStringField(jsonResponse, "ssn");
result.principal_business = extractStringField(jsonResponse, "principal_business");
result.business_code = extractStringField(jsonResponse, "business_code");
result.business_name = extractStringField(jsonResponse, "business_name");
result.business_address = extractStringField(jsonResponse, "business_address");
result.gross_income = extractNumberField(jsonResponse, "gross_income");
result.total_expenses = extractNumberField(jsonResponse, "total_expenses");
result.net_profit_loss = extractNumberField(jsonResponse, "net_profit_loss");
result.accounting_method = extractStringField(jsonResponse, "accounting_method");
result.material_participation = extractBooleanField(jsonResponse, "material_participation");
result.form_1099_required = extractBooleanField(jsonResponse, "form_1099_required");
return result;
}
private static String extractStringField(String json, String fieldName) {
String pattern = "\"" + fieldName + "\":\"";
int index = json.indexOf(pattern);
if (index == -1) {
return null;
}
int startIndex = index + pattern.length();
int endIndex = json.indexOf("\"", startIndex);
if (endIndex == -1) {
return null;
}
String value = json.substring(startIndex, endIndex);
return value.isEmpty() ? null : value.replace("\\\"", "\"");
}
private static Double extractNumberField(String json, String fieldName) {
String pattern = "\"" + fieldName + "\":";
int index = json.indexOf(pattern);
if (index == -1) {
return null;
}
int startIndex = index + pattern.length();
int endIndex = startIndex;
while (endIndex < json.length() && (Character.isDigit(json.charAt(endIndex)) || json.charAt(endIndex) == '.' || json.charAt(endIndex) == '-')) {
endIndex++;
}
if (startIndex == endIndex) {
return null;
}
try {
return Double.parseDouble(json.substring(startIndex, endIndex));
} catch (NumberFormatException e) {
return null;
}
}
private static Boolean extractBooleanField(String json, String fieldName) {
String pattern = "\"" + fieldName + "\":";
int index = json.indexOf(pattern);
if (index == -1) {
return null;
}
int startIndex = index + pattern.length();
if (json.startsWith("true", startIndex)) {
return true;
} else if (json.startsWith("false", startIndex)) {
return false;
} else if (json.startsWith("null", startIndex)) {
return null;
}
return null;
}
private static String escapeJson(String input) {
return input.replace("\\", "\\\\")
.replace("\"", "\\\"")
.replace("\n", "\\n")
.replace("\r", "\\r")
.replace("\t", "\\t");
}
public static void main(String[] args) throws IOException, InterruptedException {
if (args.length == 0) {
System.err.println("Usage: java ScheduleCProcessor <path-to-schedule-c-file>");
System.exit(1);
}
processScheduleCForm1040(args[0]);
}
}// 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/ioutil"
"net/http"
"os"
"time"
)
// ScheduleC represents the extracted Schedule C form data
type ScheduleC struct {
ProprietorName *string `json:"proprietor_name"`
SSN *string `json:"ssn"`
PrincipalBusiness *string `json:"principal_business"`
BusinessCode *string `json:"business_code"`
BusinessName *string `json:"business_name"`
BusinessAddress *string `json:"business_address"`
GrossIncome *float64 `json:"gross_income"`
TotalExpenses *float64 `json:"total_expenses"`
NetProfitLoss *float64 `json:"net_profit_loss"`
AccountingMethod *string `json:"accounting_method"`
MaterialParticipation *bool `json:"material_participation"`
Form1099Required *bool `json:"form_1099_required"`
}
// ParseRunOutput represents the output from a parse run
type ParseRunOutput struct {
Chunks []struct {
Content string `json:"content"`
} `json:"chunks"`
}
// ParseRun represents a parse run response
type ParseRun struct {
Status string `json:"status"`
Output ParseRunOutput `json:"output"`
}
// ExtractRun represents an extract run response
type ExtractRun struct {
Status string `json:"status"`
Output struct {
Value ScheduleC `json:"value"`
} `json:"output"`
}
// ProcessScheduleCForm1040 processes a Schedule C (Form 1040) document through parse + extract pipeline
func ProcessScheduleCForm1040(filePath string) (*ScheduleC, error) {
fmt.Printf("Processing Schedule C form: %s\n", filePath)
// Read local file and convert to data URL for upload
fileBuffer, err := ioutil.ReadFile(filePath)
if err != nil {
return nil, fmt.Errorf("failed to read file: %w", err)
}
base64Data := base64.StdEncoding.EncodeToString(fileBuffer)
dataURL := fmt.Sprintf("data:application/octet-stream;base64,%s", base64Data)
apiKey := os.Getenv("EXTEND_API_KEY")
if apiKey == "" {
return nil, fmt.Errorf("EXTEND_API_KEY environment variable not set")
}
// Step 1: Parse the Schedule C form to markdown with agentic OCR
fmt.Println("Step 1: Parsing Schedule C form...")
parseRun, err := createAndPollParseRun(dataURL, apiKey)
if err != nil {
return nil, err
}
if parseRun.Status != "PROCESSED" {
return nil, fmt.Errorf("parse run failed with status: %s", parseRun.Status)
}
// Log parsed markdown for debugging
parsedMarkdown := ""
for i, chunk := range parseRun.Output.Chunks {
if i > 0 {
parsedMarkdown += "\n\n"
}
parsedMarkdown += chunk.Content
}
fmt.Println("Parsed markdown preview (first 500 chars):")
if len(parsedMarkdown) > 500 {
fmt.Println(parsedMarkdown[:500] + "...\n")
} else {
fmt.Println(parsedMarkdown + "\n")
}
// Step 2: Extract structured fields with performance mode
fmt.Println("Step 2: Extracting Schedule C fields...")
extractRun, err := createAndPollExtractRun(dataURL, apiKey)
if err != nil {
return nil, err
}
if extractRun.Status != "PROCESSED" {
return nil, fmt.Errorf("extract run failed with status: %s", extractRun.Status)
}
extractedData := &extractRun.Output.Value
// Validation: Ensure financial calculations are consistent
if extractedData.GrossIncome != nil && extractedData.TotalExpenses != nil && extractedData.NetProfitLoss != nil {
calculatedNetProfit := *extractedData.GrossIncome - *extractedData.TotalExpenses
difference := calculatedNetProfit - *extractedData.NetProfitLoss
if difference < 0 {
difference = -difference
}
if difference > 1 {
// Allow $1 rounding difference
fmt.Printf("⚠️ Financial consistency check: Net profit mismatch detected. Calculated: %f, Extracted: %f\n",
calculatedNetProfit, *extractedData.NetProfitLoss)
}
}
fmt.Println("\n✅ Schedule C extraction complete:")
jsonData, _ := json.MarshalIndent(extractedData, "", " ")
fmt.Println(string(jsonData))
return extractedData, nil
}
func createAndPollParseRun(dataURL, apiKey string) (*ParseRun, error) {
requestBody := map[string]interface{}{
"file": map[string]string{
"url": dataURL,
},
"config": map[string]interface{}{
"blockOptions": map[string]interface{}{
"text": map[string]interface{}{
"agentic": map[string]bool{
"enabled": true,
},
},
},
"chunkingStrategy": map[string]string{
"type": "document",
},
},
}
body, _ := json.Marshal(requestBody)
req, _ := http.NewRequest("POST", "https://api.extend.ai/v1/parse-runs", bytes.NewBuffer(body))
req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", apiKey))
req.Header.Set("Content-Type", "application/json")
client := &http.Client{}
resp, err := client.Do(req)
if err != nil {
return nil, fmt.Errorf("failed to create parse run: %w", err)
}
defer resp.Body.Close()
var result map[string]interface{}
json.NewDecoder(resp.Body).Decode(&result)
runID := result["id"].(string)
// Poll for completion
return pollParseRun(runID, apiKey)
}
func pollParseRun(runID, apiKey string) (*ParseRun, error) {
for {
req, _ := http.NewRequest("GET", fmt.Sprintf("https://api.extend.ai/v1/parse-runs/%s", runID), nil)
req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", apiKey))
client := &http.Client{}
resp, err := client.Do(req)
if err != nil {
return nil, fmt.Errorf("failed to poll parse run: %w", err)
}
var parseRun ParseRun
json.NewDecoder(resp.Body).Decode(&parseRun)
resp.Body.Close()
if parseRun.Status == "PROCESSED" || parseRun.Status == "FAILED" {
return &parseRun, nil
}
time.Sleep(2 * time.Second)
}
}
func createAndPollExtractRun(dataURL, apiKey string) (*ExtractRun, error) {
schema := map[string]interface{}{
"type": "object",
"properties": map[string]interface{}{
"proprietor_name": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Name of the sole proprietor",
},
"ssn": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Social Security Number of the proprietor",
},
"principal_business": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Principal business or profession, including product or service",
},
"business_code": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Business code from IRS instructions",
},
"business_name": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Separate business name if applicable",
},
"business_address": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Business address including city, state, and ZIP code",
},
"gross_income": map[string]interface{}{
"type": []string{"number", "null"},
"description": "Gross income from business operations",
},
"total_expenses": map[string]interface{}{
"type": []string{"number", "null"},
"description": "Total business expenses",
},
"net_profit_loss": map[string]interface{}{
"type": []string{"number", "null"},
"description": "Net profit or loss from business",
},
"accounting_method": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Accounting method used: Cash, Accrual, or Other",
},
"material_participation": map[string]interface{}{
"type": []string{"boolean", "null"},
"description": "Whether proprietor materially participated in business operation",
},
"form_1099_required": map[string]interface{}{
"type": []string{"boolean", "null"},
"description": "Whether Form 1099 filing is required",
},
},
}
requestBody := map[string]interface{}{
"file": map[string]string{
"url": dataURL,
},
"config": map[string]interface{}{
"schema": schema,
"baseProcessor": "extraction_performance",
"advancedOptions": map[string]interface{}{
"reviewAgent": map[string]bool{
"enabled": true,
},
"advancedMultimodalEnabled": true,
},
},
}
body, _ := json.Marshal(requestBody)
req, _ := http.NewRequest("POST", "https://api.extend.ai/v1/extract-runs", bytes.NewBuffer(body))
req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", apiKey))
req.Header.Set("Content-Type", "application/json")
client := &http.Client{}
resp, err := client.Do(req)
if err != nil {
return nil, fmt.Errorf("failed to create extract run: %w", err)
}
defer resp.Body.Close()
var result map[string]interface{}
json.NewDecoder(resp.Body).Decode(&result)
runID := result["id"].(string)
// Poll for completion
return pollExtractRun(runID, apiKey)
}
func pollExtractRun(runID, apiKey string) (*ExtractRun, error) {
for {
req, _ := http.NewRequest("GET", fmt.Sprintf("https://api.extend.ai/v1/extract-runs/%s", runID), nil)
req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", apiKey))
client := &http.Client{}
resp, err := client.Do(req)
if err != nil {
return nil, fmt.Errorf("failed to poll extract run: %w", err)
}
var extractRun ExtractRun
json.NewDecoder(resp.Body).Decode(&extractRun)
resp.Body.Close()
if extractRun.Status == "PROCESSED" || extractRun.Status == "FAILED" {
return &extractRun, nil
}
time.Sleep(2 * time.Second)
}
}// Deploy the "Schedule C (Form 1040)" 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/schedule-c-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: schedule-c-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, "schedule-c-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": "Schedule C (Form 1040) 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": {
"ssn": {
"type": [
"string",
"null"
],
"description": "Social Security Number of the proprietor"
},
"gross_income": {
"type": [
"number",
"null"
],
"description": "Gross income from business operations"
},
"business_code": {
"type": [
"string",
"null"
],
"description": "Business code from IRS instructions"
},
"business_name": {
"type": [
"string",
"null"
],
"description": "Separate business name if applicable"
},
"total_expenses": {
"type": [
"number",
"null"
],
"description": "Total business expenses"
},
"net_profit_loss": {
"type": [
"number",
"null"
],
"description": "Net profit or loss from business"
},
"proprietor_name": {
"type": [
"string",
"null"
],
"description": "Name of the sole proprietor"
},
"business_address": {
"type": [
"string",
"null"
],
"description": "Business address including city, state, and ZIP code"
},
"accounting_method": {
"type": [
"string",
"null"
],
"description": "Accounting method used: Cash, Accrual, or Other"
},
"form_1099_required": {
"type": [
"boolean",
"null"
],
"description": "Whether Form 1099 filing is required"
},
"principal_business": {
"type": [
"string",
"null"
],
"description": "Principal business or profession, including product or service"
},
"material_participation": {
"type": [
"boolean",
"null"
],
"description": "Whether proprietor materially participated in business operation"
}
}
},
"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 / "schedule-c-extraction.json"
state: dict = {}
if STATE_FILE.exists():
state = json.loads(STATE_FILE.read_text())
def save_state():
STATE_DIR.mkdir(parents=True, exist_ok=True)
STATE_FILE.write_text(json.dumps(state, indent=2))
WORKFLOW = {
"name": "Schedule C (Form 1040) 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": {
"proprietor_name": {
"type": ["string", "null"],
"description": "Name of the sole proprietor",
},
"ssn": {
"type": ["string", "null"],
"description": "Social Security Number of the proprietor",
},
"principal_business": {
"type": ["string", "null"],
"description": "Principal business or profession, including product or service",
},
"business_code": {
"type": ["string", "null"],
"description": "Business code from IRS instructions",
},
"business_name": {
"type": ["string", "null"],
"description": "Separate business name if applicable",
},
"business_address": {
"type": ["string", "null"],
"description": "Business address including city, state, and ZIP code",
},
"gross_income": {
"type": ["number", "null"],
"description": "Gross income from business operations",
},
"total_expenses": {
"type": ["number", "null"],
"description": "Total business expenses",
},
"net_profit_loss": {
"type": ["number", "null"],
"description": "Net profit or loss from business",
},
"accounting_method": {
"type": ["string", "null"],
"description": "Accounting method used: Cash, Accrual, or Other",
},
"material_participation": {
"type": ["boolean", "null"],
"description": "Whether proprietor materially participated in business operation",
},
"form_1099_required": {
"type": ["boolean", "null"],
"description": "Whether Form 1099 filing is required",
},
},
},
"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 an existing workflow with the same name
try:
workflows_list = client.workflows.list(name=WORKFLOW["name"])
items = workflows_list.data if hasattr(workflows_list, "data") else []
existing = next(
(w for w in items if w.name == WORKFLOW["name"]), None
)
if existing and existing.id:
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:
# Lookup is best-effort; fall through to create
pass
if not state.get("workflowId"):
created = client.workflows.create(**WORKFLOW)
workflow_id = created.id
if not workflow_id:
raise ValueError("Could not read created workflow id from response")
state["workflowId"] = workflow_id
save_state()
print(f"+ created workflow ({workflow_id})")
# Deploy the current draft as a new version so the workflow is runnable
try:
client.workflows.create_version(id=state["workflowId"])
except Exception:
# Best-effort: some accounts/plans may not require this explicit step
pass
workflow_id = state["workflowId"]
print("\nDone. Run documents through it with:")
print(
f' POST https://api.extend.ai/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__":
try:
main()
except Exception as e:
print(str(e), file=sys.stderr)
sys.exit(1)// This code calls Extend's REST API directly using Java's built-in HttpClient.
// Extend does not publish an official Java SDK; this approach has zero external dependencies.
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 ScheduleCProvisioner {
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("schedule-c-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();
System.out.println("Deploying \"" + workflow.get("name") + "\"…");
if (state.workflowId != null && !state.workflowId.isEmpty()) {
System.out.println("✓ workflow already provisioned (" + state.workflowId + ") — updating steps");
Map<String, Object> updateBody = new LinkedHashMap<>();
updateBody.put("steps", workflow.get("steps"));
api("POST", "/workflows/" + state.workflowId, updateBody);
} else {
try {
String encodedName = URLEncoder.encode((String) workflow.get("name"), StandardCharsets.UTF_8);
Map<String, Object> listResponse = api("GET", "/workflows?name=" + encodedName, null);
List<?> items = (List<?>) listResponse.getOrDefault("data",
listResponse.getOrDefault("items", List.of()));
for (Object item : items) {
if (item instanceof Map) {
Map<?, ?> itemMap = (Map<?, ?>) item;
if (workflow.get("name").equals(itemMap.get("name")) && itemMap.get("id") != null) {
state.workflowId = (String) itemMap.get("id");
saveState();
System.out.println("✓ workflow \"" + workflow.get("name") + "\" found in your account ("
+ state.workflowId + ") — updating steps");
Map<String, Object> updateBody = new LinkedHashMap<>();
updateBody.put("steps", workflow.get("steps"));
api("POST", "/workflows/" + state.workflowId, 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", "Schedule C (Form 1040) 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);
parseConfig.put("chunkingStrategy", Map.of("type", "document"));
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("proprietor_name",
Map.of("type", List.of("string", "null"), "description", "Name of the sole proprietor"));
properties.put("ssn",
Map.of("type", List.of("string", "null"), "description", "Social Security Number of the proprietor"));
properties.put("principal_business", Map.of("type", List.of("string", "null"),
"description", "Principal business or profession, including product or service"));
properties.put("business_code",
Map.of("type", List.of("string", "null"), "description", "Business code from IRS instructions"));
properties.put("business_name",
Map.of("type", List.of("string", "null"), "description", "Separate business name if applicable"));
properties.put("business_address", Map.of("type", List.of("string", "null"),
"description", "Business address including city, state, and ZIP code"));
properties.put("gross_income",
Map.of("type", List.of("number", "null"), "description", "Gross income from business operations"));
properties.put("total_expenses",
Map.of("type", List.of("number", "null"), "description", "Total business expenses"));
properties.put("net_profit_loss",
Map.of("type", List.of("number", "null"), "description", "Net profit or loss from business"));
properties.put("accounting_method", Map.of("type", List.of("string", "null"),
"description", "Accounting method used: Cash, Accrual, or Other"));
properties.put("material_participation", Map.of("type", List.of("boolean", "null"),
"description", "Whether proprietor materially participated in business operation"));
properties.put("form_1099_required", Map.of("type", List.of("boolean", "null"),
"description", "Whether Form 1099 filing is required"));
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<>();
advancedOptions.put("reviewAgent", Map.of("enabled", true));
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> api(String method, String pathName, Map<String, Object> body)
throws IOException, InterruptedException {
String url = API + pathName;
HttpRequest.Builder requestBuilder = HttpRequest.newBuilder()
.uri(URI.create(url))
.header("Authorization", "Bearer " + API_KEY)
.header("x-extend-api-version", VERSION);
if (body != null) {
String jsonBody = mapToJson(body);
requestBuilder.header("Content-Type", "application/json")
.method(method, HttpRequest.BodyPublishers.ofString(jsonBody));
} else {
requestBuilder.method(method, HttpRequest.BodyPublishers.noBody());
}
HttpRequest request = requestBuilder.build();
HttpResponse<String> response = HTTP_CLIENT.send(request, HttpResponse.BodyHandlers.ofString());
Map<String, Object> data = jsonToMap(response.body());
if (response.statusCode() < 200 || response.statusCode() >= 300) {
String errorMsg = mapToJson(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> stateMap = jsonToMap(content);
if (stateMap.containsKey("workflowId")) {
state.workflowId = (String) stateMap.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);
}
String json = mapToJson(stateMap);
Files.writeString(STATE_FILE, json);
}
private static String mapToJson(Map<String, Object> map) {
StringBuilder sb = new StringBuilder();
sb.append("{");
boolean first = true;
for (Map.Entry<String, Object> entry : map.entrySet()) {
if (!first) sb.append(",");
first = false;
sb.append("\"").append(escapeJson(entry.getKey())).append("\":");
sb.append(valueToJson(entry.getValue()));
}
sb.append("}");
return sb.toString();
}
private static String valueToJson(Object value) {
if (value == null) {
return "null";
} else if (value instanceof String) {
return "\"" + escapeJson((String) value) + "\"";
} else if (value instanceof Number) {
return value.toString();
} else if (value instanceof Boolean) {
return value.toString();
} else if (value instanceof Map) {
return mapToJson((Map<String, Object>) value);
} else if (value instanceof List) {
StringBuilder sb = new StringBuilder("[");
List<?> list = (List<?>) value;
for (int i = 0; i < list.size(); i++) {
if (i > 0) sb.append(",");
sb.append(valueToJson(list.get(i)));
}
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");
}
private static Map<String, Object> jsonToMap(String json) {
json = json.trim();
if (!json.startsWith("{")) {
return new LinkedHashMap<>();
}
Map<String, Object> map = new LinkedHashMap<>();
int depth = 0;
int i = 1;
String key = null;
StringBuilder currentValue = new StringBuilder();
while (i < json.length()) {
char c = json.charAt(i);
if (c == '"' && (i == 0 || json.charAt(i - 1) != '\\')) {
int endQuote = i + 1;
while (endQuote < json.length()) {
if (json.charAt(endQuote) == '"' && json.charAt(endQuote - 1) != '\\') {
break;
}
endQuote++;
}
String quoted = json.substring(i + 1, endQuote);
i = endQuote;
if (key == null) {
key = unescapeJson(quoted);
while (i < json.length() && (json.charAt(i) == ':' || Character.isWhitespace(json.charAt(i)))) {
i++;
}
i--;
} else {
map.put(key, unescapeJson(quoted));
key = null;
}
} else if ((c == '{' || c == '[') && (i == 0 || json.charAt(i - 1) != '\\')) {
depth++;
currentValue.append(c);
} else if ((c == '}' || c == ']') && (i == 0 || json.charAt(i - 1) != '\\')) {
depth--;
if (depth == 0 && c == '}') {
break;
}
currentValue.append(c);
} else if (c == ',' && depth == 0) {
if (key != null && currentValue.length() > 0) {
String val = currentValue.toString().trim();
map.put(key, parseJsonValue(val));
key = null;
currentValue = new StringBuilder();
}
} else if (!Character.isWhitespace(c) || depth > 0) {
currentValue.append(c);
}
i++;
}
if (key != null && currentValue.length() > 0) {
String val = currentValue.toString().trim();
map.put(key, parseJsonValue(val));
}
return map;
}
private static Object parseJsonValue(String val) {
val = val.trim();
if (val.equals("null")) {
return null;
} else if (val.equals("true")) {
return true;
} else if (val.equals("false")) {
return false;
} else if (val.startsWith("{")) {
return jsonToMap(val);
} else if (val.startsWith("[")) {
return jsonToList(val);
} else {
try {
if (val.contains(".")) {
return Double.parseDouble(val);
} else {
return Long.parseLong(val);
}
} catch (NumberFormatException e) {
return val;
}
}
}
private static List<?> jsonToList(String json) {
List<Object> list = new java.util.ArrayList<>();
json = json.trim();
if (!json.startsWith("[") || !json.endsWith("]")) {
return list;
}
json = json.substring(1, json.length() - 1).trim();
if (json.isEmpty()) {
return list;
}
int depth = 0;
StringBuilder current = new StringBuilder();
for (int i = 0; i < json.length(); i++) {
char c = json.charAt(i);
if ((c == '{' || c == '[') && (i == 0 || json.charAt(i - 1) != '\\')) {
depth++;
current.append(c);
} else if ((c == '}' || c == ']') && (i == 0 || json.charAt(i - 1) != '\\')) {
depth--;
current.append(c);
} else if (c == ',' && depth == 0) {
list.add(parseJsonValue(current.toString()));
current = new StringBuilder();
} else {
current.append(c);
}
}
if (current.length() > 0) {
list.add(parseJsonValue(current.toString()));
}
return list;
}
private static String unescapeJson(String s) {
return s.replace("\\\"", "\"")
.replace("\\\\", "\\")
.replace("\\n", "\n")
.replace("\\r", "\r")
.replace("\\t", "\t");
}
}// This code uses the Extend REST API directly because Extend has no official Go SDK yet.
// It deploys the "Schedule C (Form 1040)" 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: schedule-c-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)
}
wd, err := os.Getwd()
if err != nil {
fmt.Fprintf(os.Stderr, "Failed to get working directory: %v\n", err)
os.Exit(1)
}
stateDir = filepath.Join(wd, ".extend")
stateFile = filepath.Join(stateDir, "schedule-c-extraction.json")
if data, err := os.ReadFile(stateFile); err == nil {
json.Unmarshal(data, &state)
}
}
func saveState() error {
if err := os.MkdirAll(stateDir, 0755); err != nil {
return err
}
data, err := json.MarshalIndent(state, "", " ")
if err != nil {
return err
}
return os.WriteFile(stateFile, data, 0644)
}
func apiCall(method, pathName string, body interface{}) (map[string]interface{}, error) {
var reqBody io.Reader
if body != nil {
data, err := json.Marshal(body)
if err != nil {
return nil, err
}
reqBody = bytes.NewReader(data)
}
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")
}
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
}
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
}
var workflow = map[string]interface{}{
"name": "Schedule C (Form 1040) 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{}{
"proprietor_name": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Name of the sole proprietor",
},
"ssn": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Social Security Number of the proprietor",
},
"principal_business": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Principal business or profession, including product or service",
},
"business_code": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Business code from IRS instructions",
},
"business_name": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Separate business name if applicable",
},
"business_address": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Business address including city, state, and ZIP code",
},
"gross_income": map[string]interface{}{
"type": []string{"number", "null"},
"description": "Gross income from business operations",
},
"total_expenses": map[string]interface{}{
"type": []string{"number", "null"},
"description": "Total business expenses",
},
"net_profit_loss": map[string]interface{}{
"type": []string{"number", "null"},
"description": "Net profit or loss from business",
},
"accounting_method": map[string]interface{}{
"type": []string{"string", "null"},
"description": "Accounting method used: Cash, Accrual, or Other",
},
"material_participation": map[string]interface{}{
"type": []string{"boolean", "null"},
"description": "Whether proprietor materially participated in business operation",
},
"form_1099_required": map[string]interface{}{
"type": []string{"boolean", "null"},
"description": "Whether Form 1099 filing is required",
},
},
},
"baseProcessor": "extraction_performance",
"advancedOptions": map[string]interface{}{
"reviewAgent": map[string]interface{}{
"enabled": true,
},
"advancedMultimodalEnabled": true,
},
},
},
},
},
}
func main() {
workflowName := workflow["name"].(string)
fmt.Printf("Deploying \"%s\"…\n", workflowName)
if state.WorkflowID != "" {
fmt.Printf("✓ workflow already provisioned (%s) — updating steps\n", state.WorkflowID)
_, err := apiCall("POST", "/workflows/"+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 := "/workflows?name=" + url.QueryEscape(workflowName)
list, err := apiCall("GET", listPath, nil)
if err == nil {
var items []map[string]interface{}
if data, ok := list["data"].([]interface{}); ok {
for _, item := range data {
items = append(items, item.(map[string]interface{}))
}
} else if data, ok := list["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", "/workflows/"+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", "/workflows/"+state.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, state.WorkflowID)
fmt.Println("Or open the workflow in the Extend dashboard to review and deploy it.")
}Schedule C (Form 1040) is used to report profit or loss from a sole proprietorship business. This template captures business identification details, income sources, and itemized business expenses required for individual tax filing.