QuickBooks Online API Expert Guide

Overview

The QuickBooks Online API provides comprehensive access to accounting data and operations for QuickBooks Online companies. This skill enables you to build integrations that handle invoicing, payments, customer management, inventory tracking, and financial reporting. The API uses OAuth 2.0 for authentication and supports operations across all major accounting entities including customers, invoices, payments, items, accounts, and more.

The QuickBooks Online API is REST-based, returns JSON or XML responses, and provides SDKs for Java, Python, PHP, Node.js, and C#. It supports both sandbox (development) and production environments.

When to Use This Skill

Use this skill when:

• Building QuickBooks integrations for accounting automation

• Implementing invoicing workflows or payment processing

• Creating customer or vendor management features

• Working with QuickBooks Online API authentication (OAuth2)

• Troubleshooting API errors or validation failures

• Implementing batch operations for bulk data updates

• Setting up change data capture (CDC) or webhooks for data synchronization

• Designing multi-currency or international accounting integrations

• Building reports or analytics on top of QuickBooks data

• Migrating data to/from QuickBooks Online

Authentication & OAuth2 Setup

OAuth 2.0 Flow

QuickBooks Online API requires OAuth 2.0 authentication. The flow involves:

• Register your app at developer.intuit.com to get Client ID and Client Secret

• Direct users to authorization URL where they grant access to their QuickBooks company

• Exchange authorization code for tokens (access token + refresh token)

• Use access token in API requests (Authorization: Bearer header)

• Refresh tokens before expiration to maintain access

Token Lifecycle

Access Tokens:

• Valid for 3600 seconds (1 hour)

• Include in Authorization header: Authorization: Bearer {access_token}

• Return 401 Unauthorized when expired

Refresh Tokens:

• Valid for 100 days from issuance

• Use to obtain new access token + refresh token pair

• Previous refresh token expires 24 hours after new one is issued

• Always use the most recent refresh token

Token Refresh Pattern

Node.js Example:

const oauthClient = require('intuit-oauth');

// Refresh access token oauthClient.refresh() .then(function(authResponse) { const newAccessToken = authResponse.token.access_token; const newRefreshToken = authResponse.token.refresh_token; const expiresIn = authResponse.token.expires_in; // 3600 seconds

// Store new tokens securely (database, encrypted storage)
console.log('Tokens refreshed successfully');

}) .catch(function(e) { console.error('Token refresh failed:', e.originalMessage); // Handle re-authentication if refresh token is invalid });

Python Example:

from intuitlib.client import AuthClient

auth_client = AuthClient( client_id='YOUR_CLIENT_ID', client_secret='YOUR_CLIENT_SECRET', redirect_uri='YOUR_REDIRECT_URI', environment='sandbox' # or 'production' )

Refresh tokens

auth_client.refresh(refresh_token='STORED_REFRESH_TOKEN')

Get new tokens

new_access_token = auth_client.access_token new_refresh_token = auth_client.refresh_token

Best Practices

• Refresh proactively: Refresh tokens before they expire (e.g., after 50 minutes)

• Store securely: Encrypt tokens in database, never commit to version control

• Handle 401 responses: Automatically attempt token refresh on authentication errors

• Realm ID (Company ID): Store the realmId returned during OAuth - required for all API calls

• Scopes: Request only necessary scopes (accounting, payments, etc.)

Core Entities Reference

Customer

Represents customers and sub-customers (jobs) in QuickBooks.

Key Fields:

• Id (string, read-only): Unique identifier

• DisplayName (string, required): Customer display name (must be unique)

• GivenName , FamilyName (string): First and last name

• CompanyName (string): Company name for business customers

• PrimaryEmailAddr (object): Email address { "Address": "email@example.com" }

• PrimaryPhone (object): Phone number { "FreeFormNumber": "(555) 123-4567" }

• BillAddr , ShipAddr (object): Billing and shipping addresses

• Balance (decimal, read-only): Current outstanding balance

• Active (boolean): Whether customer is active

• SyncToken (string, required for updates): Version number for optimistic locking

Reference Type: Use CustomerRef in transactions: { "value": "123", "name": "Customer Name" }

Invoice

Represents sales invoices sent to customers.

Key Fields:

• Id (string, read-only): Unique identifier

• DocNumber (string): Invoice number (auto-generated if not provided)

• TxnDate (date): Transaction date (YYYY-MM-DD format)

• DueDate (date): Payment due date

• CustomerRef (object, required): Reference to customer { "value": "customerId" }

• Line (array, required): Invoice line items (see Line Items section)

• TotalAmt (decimal, read-only): Calculated total amount

• Balance (decimal, read-only): Remaining unpaid balance

• EmailStatus (enum): NotSet, NeedToSend, EmailSent

• BillEmail (object): Customer email for invoice delivery

• TxnTaxDetail (object): Tax calculation details

• LinkedTxn (array): Linked transactions (payments, credit memos)

• SyncToken (string, required for updates): Version number

Line Items:

{ "Line": [ { "Amount": 100.00, "DetailType": "SalesItemLineDetail", "SalesItemLineDetail": { "ItemRef": { "value": "1", "name": "Services" }, "Qty": 1, "UnitPrice": 100.00, "TaxCodeRef": { "value": "TAX" } } }, { "Amount": 100.00, "DetailType": "SubTotalLineDetail", "SubTotalLineDetail": {} } ] }

Payment

Represents payments received from customers against invoices.

Key Fields:

• Id (string, read-only): Unique identifier

• TotalAmt (decimal, required): Total payment amount

• CustomerRef (object, required): Reference to customer

• PaymentMethodRef (object): Payment method (cash, check, credit card, etc.)

• PaymentRefNum (string): Reference number (check number, transaction ID)

• TxnDate (date): Payment date

• DepositToAccountRef (object): Bank account for deposit

• Line (array): Payment application to invoices/credit memos

• UnappliedAmt (decimal, read-only): Amount not applied to invoices

• SyncToken (string, required for updates): Version number

Payment Line Item (applies payment to invoice):

{ "Line": [ { "Amount": 100.00, "LinkedTxn": [ { "TxnId": "123", "TxnType": "Invoice" } ] } ] }

Item

Represents products or services sold.

Types:

• Service : Services (consulting, labor, etc.)

• Inventory : Physical products tracked in inventory

• NonInventory : Physical products not tracked

• Category : Grouping for other items

Key Fields:

• Id (string, read-only): Unique identifier

• Name (string, required): Item name (must be unique)

• Type (enum, required): Service, Inventory, NonInventory, Category

• Description (string): Item description

• UnitPrice (decimal): Sales price

• PurchaseCost (decimal): Purchase/cost price

• IncomeAccountRef (object, required): Income account reference

• ExpenseAccountRef (object): Expense account for purchases

• TrackQtyOnHand (boolean): Whether to track inventory quantity

• QtyOnHand (decimal): Current inventory quantity

• Active (boolean): Whether item is active

Account

Represents accounts in the chart of accounts.

Key Fields:

• Id (string, read-only): Unique identifier

• Name (string, required): Account name

• AccountType (enum, required): Bank, Accounts Receivable, Accounts Payable, Income, Expense, etc.

• AccountSubType (enum): More specific type (CashOnHand, Checking, Savings, etc.)

• CurrentBalance (decimal, read-only): Current account balance

• Active (boolean): Whether account is active

• Classification (enum): Asset, Liability, Equity, Revenue, Expense

Common Account Types:

• Bank : Bank and cash accounts

• Accounts Receivable : Customer balances

• Accounts Payable : Vendor balances

• Income : Revenue accounts

• Expense : Expense accounts

• Other Current Asset : Short-term assets

• Fixed Asset : Long-term assets

CRUD Operations Patterns

Create Operations

Minimum Required Fields: Each entity has specific required fields (usually a name/reference and amount).

Endpoint Pattern: POST /v3/company/{realmId}/{entityName}

Request Headers:

Authorization: Bearer {access_token} Accept: application/json Content-Type: application/json

Python Example - Create Invoice:

import requests

realm_id = "YOUR_REALM_ID" access_token = "YOUR_ACCESS_TOKEN"

url = f"https://sandbox-quickbooks.api.intuit.com/v3/company/{realm_id}/invoice"

headers = { "Authorization": f"Bearer {access_token}", "Accept": "application/json", "Content-Type": "application/json" }

invoice_data = { "Line": [ { "Amount": 100.00, "DetailType": "SalesItemLineDetail", "SalesItemLineDetail": { "ItemRef": {"value": "1"} } } ], "CustomerRef": {"value": "1"} }

response = requests.post(url, json=invoice_data, headers=headers)

if response.status_code == 200: invoice = response.json()['Invoice'] print(f"Invoice created: {invoice['Id']}") else: print(f"Error: {response.status_code} - {response.text}")

Read Operations

Single Entity: GET /v3/company/{realmId}/{entityName}/{entityId}

Node.js Example - Read Customer:

const axios = require('axios');

async function readCustomer(realmId, customerId, accessToken) { const url = https://sandbox-quickbooks.api.intuit.com/v3/company/${realmId}/customer/${customerId};

try { const response = await axios.get(url, { headers: { 'Authorization': Bearer ${accessToken}, 'Accept': 'application/json' } });

return response.data.Customer;

} catch (error) { if (error.response && error.response.status === 401) { // Token expired, refresh and retry console.error('Authentication failed - refresh token needed'); } else { console.error('Read failed:', error.response?.data || error.message); } throw error; } }

Update Operations

Two types of updates:

1. Full Update: All writable fields must be included. Omitted fields are set to NULL.

2. Sparse Update: Only specified fields are updated. Set "sparse": true in request body.

Important: Always include SyncToken from the latest read response. This prevents concurrent modification conflicts.

Python Example - Sparse Update Customer Email:

import requests

def sparse_update_customer(realm_id, customer_id, sync_token, new_email, access_token): url = f"https://sandbox-quickbooks.api.intuit.com/v3/company/{realm_id}/customer"

headers = {
    "Authorization": f"Bearer {access_token}",
    "Accept": "application/json",
    "Content-Type": "application/json"
}

# Sparse update - only updating email
customer_data = {
    "Id": customer_id,
    "SyncToken": sync_token,
    "sparse": True,
    "PrimaryEmailAddr": {
        "Address": new_email
    }
}

response = requests.post(url, json=customer_data, headers=headers)

if response.status_code == 200:
    updated_customer = response.json()['Customer']
    print(f"Customer updated, new SyncToken: {updated_customer['SyncToken']}")
    return updated_customer
else:
    print(f"Update failed: {response.text}")
    return None

SyncToken Handling:

1. Read entity to get latest SyncToken

customer = read_customer(realm_id, customer_id, access_token)

2. Update with current SyncToken

updated = sparse_update_customer( realm_id, customer_id, customer['SyncToken'], # Use current sync token "newemail@example.com", access_token )

3. Store new SyncToken for next update

new_sync_token = updated['SyncToken']

Delete Operations

Most entities use soft delete (setting Active to false) or void operations.

Soft Delete Pattern:

// Mark customer as inactive const deleteCustomer = { Id: customerId, SyncToken: currentSyncToken, sparse: true, Active: false };

// POST to update endpoint axios.post(${baseUrl}/customer, deleteCustomer, { headers });

Hard Delete (limited entities): POST /v3/company/{realmId}/{entityName}?operation=delete

{ "Id": "123", "SyncToken": "2" }

Query Language & Filtering

QuickBooks uses SQL-like query syntax with limitations.

Query Syntax

Basic Pattern:

SELECT * FROM {EntityName} WHERE {field} {operator} '{value}'

Endpoint: GET /v3/company/{realmId}/query?query={sqlQuery}

Operators

• = : Equals

• < , > , <= , >= : Comparison

• IN : Match any value in list

• LIKE : Pattern matching (only % wildcard supported, no _ )

Examples

Query customers by name:

SELECT * FROM Customer WHERE DisplayName LIKE 'Acme%'

Query invoices by date range:

SELECT * FROM Invoice WHERE TxnDate >= '2024-01-01' AND TxnDate <= '2024-12-31'

Query with ordering:

SELECT * FROM Customer WHERE Active = true ORDERBY DisplayName

Pagination:

SELECT * FROM Invoice STARTPOSITION 1 MAXRESULTS 100

Python Example - Query with Filters:

import requests from urllib.parse import quote

def query_invoices_by_customer(realm_id, customer_id, access_token): query = f"SELECT * FROM Invoice WHERE CustomerRef = '{customer_id}' ORDERBY TxnDate DESC" encoded_query = quote(query)

url = f"https://sandbox-quickbooks.api.intuit.com/v3/company/{realm_id}/query?query={encoded_query}"

headers = {
    "Authorization": f"Bearer {access_token}",
    "Accept": "application/json"
}

response = requests.get(url, headers=headers)

if response.status_code == 200:
    result = response.json()['QueryResponse']
    invoices = result.get('Invoice', [])
    print(f"Found {len(invoices)} invoices")
    return invoices
else:
    print(f"Query failed: {response.text}")
    return []

Query Limitations

• No wildcards except %: LIKE only supports % (not _ )

• No JOIN operations: Query single entity at a time

• Limited functions: No aggregate functions (SUM, COUNT, etc.)

• Max 1000 results: Use pagination for larger result sets

• All fields returned: Cannot select specific fields (always returns all)

Pagination Pattern

def query_all_customers(realm_id, access_token): all_customers = [] start_position = 1 max_results = 1000

while True:
    query = f"SELECT * FROM Customer STARTPOSITION {start_position} MAXRESULTS {max_results}"
    encoded_query = quote(query)
    url = f"{base_url}/company/{realm_id}/query?query={encoded_query}"

    response = requests.get(url, headers={"Authorization": f"Bearer {access_token}"})
    result = response.json()['QueryResponse']

    customers = result.get('Customer', [])
    if not customers:
        break

    all_customers.extend(customers)

    # Check if more results exist
    if len(customers) < max_results:
        break

    start_position += max_results

return all_customers

Batch Operations

Batch operations allow multiple API calls in a single HTTP request (up to 30 operations).

Batch Request Structure

Endpoint: POST /v3/company/{realmId}/batch

Request Body:

{ "BatchItemRequest": [ { "bId": "bid1", "operation": "create", "Customer": { "DisplayName": "New Customer 1" } }, { "bId": "bid2", "operation": "update", "Invoice": { "Id": "123", "SyncToken": "1", "sparse": true, "EmailStatus": "NeedToSend" } }, { "bId": "bid3", "operation": "query", "Query": "SELECT * FROM Customer WHERE Active = true MAXRESULTS 10" } ] }

Batch ID Tracking

Each operation has a unique bId (batch ID) for tracking results:

Response Structure:

{ "BatchItemResponse": [ { "bId": "bid1", "Customer": { "Id": "456", "DisplayName": "New Customer 1" } }, { "bId": "bid2", "Invoice": { "Id": "123", "SyncToken": "2" } }, { "bId": "bid3", "QueryResponse": { "Customer": [...] } } ] }

Node.js Example - Batch Update Customers

async function batchUpdateCustomers(realmId, customers, accessToken) { const batchItems = customers.map((customer, index) => ({ bId: customer_${index}, operation: 'update', Customer: { Id: customer.Id, SyncToken: customer.SyncToken, sparse: true, Active: true // Reactivate all customers } }));

const url = https://sandbox-quickbooks.api.intuit.com/v3/company/${realmId}/batch;

try { const response = await axios.post(url, { BatchItemRequest: batchItems }, { headers: { 'Authorization': Bearer ${accessToken}, 'Content-Type': 'application/json' } });

const results = response.data.BatchItemResponse;

// Process results by batch ID
results.forEach(result => {
  if (result.Fault) {
    console.error(`Error for ${result.bId}:`, result.Fault);
  } else {
    console.log(`Success for ${result.bId}: Customer ${result.Customer.Id}`);
  }
});

return results;

} catch (error) { console.error('Batch operation failed:', error.response?.data || error.message); throw error; } }

Benefits of Batch Operations

• Reduced API calls: 30 operations in one request vs 30 separate requests

• Lower latency: Single round-trip instead of multiple

• Rate limit friendly: Counts as single API call for rate limiting

• Atomic per operation: Each operation succeeds or fails independently

Batch Operation Types

• create : Create new entity

• update : Update existing entity

• delete : Delete entity

• query : Execute query

Error Handling & Troubleshooting

HTTP Status Codes

• 200 OK: Request successful (but may contain <Fault> element in body)

• 400 Bad Request: Invalid syntax or malformed request

• 401 Unauthorized: Invalid/expired access token

• 403 Forbidden: Insufficient permissions or restricted resource

• 404 Not Found: Resource doesn't exist

• 429 Too Many Requests: Rate limit exceeded

• 500 Internal Server Error: Server-side issue (retry once)

• 503 Service Unavailable: Service temporarily unavailable (retry with backoff)

Fault Types

Even with 200 OK, response may contain fault element:

{ "Fault": { "Error": [ { "Message": "Duplicate Name Exists Error", "Detail": "The name supplied already exists.", "code": "6240", "element": "Customer.DisplayName" } ], "type": "ValidationFault" }, "time": "2024-12-09T10:30:00.000-08:00" }

Fault Types:

ValidationFault: Invalid request data or business rule violation

• Fix: Correct request payload, check required fields

SystemFault: Server-side error

• Fix: Retry request, contact support if persists

AuthenticationFault: Invalid credentials

• Fix: Refresh access token, re-authenticate

AuthorizationFault: Insufficient permissions

• Fix: Check OAuth scopes, ensure user has admin access

Common Error Codes

Code Error Solution

6000 Business validation error Check TotalAmt and required fields

3200 Stale object (SyncToken mismatch) Re-read entity to get latest SyncToken

3100 Invalid reference Verify referenced entity exists (CustomerRef, ItemRef)

6240 Duplicate name Use unique DisplayName for Customer/Item

610 Object not found Check entity ID exists

4001 Invalid token Refresh access token

Exception Handling by SDK

Java SDK Exceptions:

• ValidationException : Validation faults

• ServiceException : Service faults

• AuthenticationException : Authentication faults

• BadRequestException : 400 status

• InvalidTokenException : 401 status

• InternalServiceException : 500 status

Python Exception Handling:

from intuitlib.exceptions import AuthClientError

try: response = requests.post(url, json=data, headers=headers) response.raise_for_status()

# Check for fault in response body
result = response.json()
if 'Fault' in result:
    fault = result['Fault']
    print(f"Fault Type: {fault['type']}")
    for error in fault['Error']:
        print(f"  Code {error['code']}: {error['Message']}")
        print(f"  Element: {error.get('element', 'N/A')}")
    return None

return result

except requests.exceptions.HTTPError as e: if e.response.status_code == 401: # Token expired, refresh print("Token expired, refreshing...") # Implement token refresh logic elif e.response.status_code == 429: # Rate limited, implement backoff print("Rate limited, backing off...") else: print(f"HTTP Error: {e.response.status_code}") print(f"Response: {e.response.text}")

except AuthClientError as e: print(f"Auth error: {str(e)}")

Debugging Strategies

• Check response body even with 200: Fault elements can appear in successful responses

• Log intuit_tid: Include in support requests for faster resolution

• Validate SyncToken: Always use latest version from read operations

• Test in sandbox first: Use sandbox companies for development

• Implement retry logic: Exponential backoff for 500/503 errors

• Parse error details: Check error.code , element , message fields

Retry Pattern with Exponential Backoff

async function apiCallWithRetry(apiFunction, maxRetries = 3) { for (let attempt = 0; attempt < maxRetries; attempt++) { try { return await apiFunction(); } catch (error) { const status = error.response?.status;

  // Retry on server errors
  if (status >= 500 && status < 600 && attempt < maxRetries - 1) {
    const delay = Math.pow(2, attempt) * 1000; // 1s, 2s, 4s
    console.log(`Attempt ${attempt + 1} failed, retrying in ${delay}ms...`);
    await new Promise(resolve => setTimeout(resolve, delay));
    continue;
  }

  // Don't retry on client errors
  throw error;
}

} }

Change Detection & Webhooks

Change Data Capture (CDC)

CDC returns entities that changed within a specified timeframe (up to 30 days).

Endpoint: GET /v3/company/{realmId}/cdc?entities={entityList}&changedSince={dateTime}

Parameters:

• entities : Comma-separated list (e.g., "Invoice,Customer,Payment")

• changedSince : ISO 8601 timestamp (e.g., "2024-12-01T09:00:00-07:00")

Python Example:

from datetime import datetime, timedelta from urllib.parse import urlencode

def get_changed_entities(realm_id, entity_types, since_datetime, access_token): # Format: 2024-12-01T09:00:00-07:00 changed_since = since_datetime.strftime('%Y-%m-%dT%H:%M:%S-07:00')

params = {
    'entities': ','.join(entity_types),
    'changedSince': changed_since
}

url = f"https://sandbox-quickbooks.api.intuit.com/v3/company/{realm_id}/cdc"

response = requests.get(
    url,
    params=params,
    headers={"Authorization": f"Bearer {access_token}"}
)

if response.status_code == 200:
    cdc_response = response.json()['CDCResponse']

    # Process changed entities
    for query_response in cdc_response:
        entity_type = query_response.get('QueryResponse', [{}])[0]

        for entity_name, entities in entity_type.items():
            if entities:
                for entity in entities:
                    status = entity.get('status', 'Updated')
                    if status == 'Deleted':
                        print(f"Deleted {entity_name}: {entity['Id']}")
                    else:
                        print(f"Changed {entity_name}: {entity['Id']}")

    return cdc_response
else:
    print(f"CDC request failed: {response.text}")
    return None

Usage: Get all invoices and customers changed in last 24 hours

since = datetime.now() - timedelta(hours=24) changes = get_changed_entities(realm_id, ['Invoice', 'Customer'], since, access_token)

Response Structure:

{ "CDCResponse": [ { "QueryResponse": [ { "Invoice": [ { "Id": "123", "MetaData": { "LastUpdatedTime": "2024-12-09T10:30:00-08:00" }, "TotalAmt": 100.00, "Balance": 50.00 // ... full invoice object } ] } ] }, { "QueryResponse": [ { "Customer": [ { "Id": "456", "status": "Deleted" } ] } ] } ], "time": "2024-12-09T11:00:00.000-08:00" }

CDC Best Practices

• Query shorter periods: Max 1000 entities per response, use hourly/daily checks

• Store last sync time: Track LastUpdatedTime to set changedSince parameter

• Handle deletes: Entities with status: "Deleted" only contain ID

• Fetch full entity: CDC returns full payload (not just changes)

• Combine with webhooks: Use webhooks for real-time, CDC as backup

Webhooks (Real-time Notifications)

Webhooks send HTTP POST notifications when data changes.

Setup:

• Configure webhook URL in developer dashboard

• Implement POST endpoint to receive notifications

• Return 200 OK within 1 second

• Process notification asynchronously

Notification Payload:

{ "eventNotifications": [ { "realmId": "123456789", "dataChangeEvent": { "entities": [ { "name": "Invoice", "id": "145", "operation": "Create", "lastUpdated": "2024-12-09T10:30:00.000Z" }, { "name": "Payment", "id": "456", "operation": "Update", "lastUpdated": "2024-12-09T10:31:00.000Z" }, { "name": "Customer", "id": "789", "operation": "Merge", "lastUpdated": "2024-12-09T10:32:00.000Z", "deletedId": "788" } ] } } ] }

Node.js Webhook Handler:

const express = require('express'); const crypto = require('crypto');

const app = express(); app.use(express.json());

// Webhook endpoint app.post('/webhooks/quickbooks', async (req, res) => { // Verify webhook signature (recommended) const signature = req.headers['intuit-signature']; const payload = JSON.stringify(req.body);

// Return 200 immediately (process async) res.status(200).send('OK');

// Process notifications asynchronously processWebhook(req.body).catch(console.error); });

async function processWebhook(notification) { for (const event of notification.eventNotifications) { const realmId = event.realmId;

for (const entity of event.dataChangeEvent.entities) {
  console.log(`${entity.operation} on ${entity.name} ID ${entity.id}`);

  // Fetch full entity data
  if (entity.operation !== 'Delete') {
    await fetchAndProcessEntity(realmId, entity.name, entity.id);
  } else {
    await handleEntityDeletion(realmId, entity.name, entity.id);
  }
}

} }

async function fetchAndProcessEntity(realmId, entityType, entityId) { // Fetch full entity using read endpoint const url = https://quickbooks.api.intuit.com/v3/company/${realmId}/${entityType.toLowerCase()}/${entityId}; // ... implement fetch and processing logic }

Webhook vs CDC Decision Matrix

Use Case Recommendation

Real-time sync Webhooks

Periodic sync (hourly/daily) CDC

Initial data load CDC

Reconnection after downtime CDC

High-volume changes CDC (reduces notification overhead)

Low-latency requirements Webhooks

Backup/redundancy Both (webhooks primary, CDC backup)

Combined Approach Pattern

class QuickBooksSync: def init(self): self.last_cdc_sync = self.load_last_sync_time()

def handle_webhook(self, notification):
    """Process real-time webhook"""
    for entity in notification['dataChangeEvent']['entities']:
        self.process_entity_change(entity)

    # Update last known change time
    self.last_cdc_sync = datetime.now()
    self.save_last_sync_time()

def periodic_cdc_sync(self):
    """Catch any missed changes"""
    changes = get_changed_entities(
        self.realm_id,
        ['Invoice', 'Customer', 'Payment'],
        self.last_cdc_sync,
        self.access_token
    )

    for entity in self.extract_entities(changes):
        if not self.entity_exists_locally(entity):
            # Missed by webhook, process now
            self.process_entity_change(entity)

    self.last_cdc_sync = datetime.now()
    self.save_last_sync_time()

Best Practices

Performance Optimization

Use batch operations for bulk changes

• Combine up to 30 operations in single request

• Reduces API calls and improves throughput

• Example: Batch update 30 customers vs 30 individual updates

Implement CDC or webhooks for syncing

• Avoid polling all entities repeatedly

• CDC returns only changed entities

• Webhooks provide real-time notifications without polling

Sparse updates minimize payload

• Only send fields being changed

• Reduces data transfer and processing time

• Prevents accidental field overwrites

Cache reference data locally

• Payment methods, tax codes, accounts rarely change

• Query once and cache with TTL

• Reduces redundant API calls

Paginate large result sets

• Use MAXRESULTS to limit query results

• Process in batches to avoid memory issues

• Example: Query 100 customers at a time

Data Integrity

Always use SyncToken for updates

• Prevents concurrent modification conflicts

• Read entity before update to get latest token

• Handle 3200 errors by re-reading and retrying

Handle concurrent modifications gracefully

def safe_update(realm_id, customer_id, changes, access_token): max_attempts = 3 for attempt in range(max_attempts): # Read latest version customer = read_customer(realm_id, customer_id, access_token)

    # Apply changes
    customer.update(changes)
    customer['sparse'] = True

    # Attempt update
    try:
        return update_customer(realm_id, customer, access_token)
    except SyncTokenError:
        if attempt == max_attempts - 1:
            raise
        continue  # Retry with fresh SyncToken

- Validate required fields before API calls

• Check business rules locally first

• Reduces validation errors from API

• Example: Verify customer exists before creating invoice

Use webhooks + CDC for reliable tracking

• Webhooks for real-time updates

• Periodic CDC as backup for missed changes

• Store last sync timestamp

Token Management

Access tokens expire after 3600 seconds

• Set up automatic refresh before expiration

• Refresh at 50-minute mark to be safe

Refresh tokens proactively

class TokenManager { constructor() { this.refreshTimer = null; }

scheduleRefresh(expiresIn) { // Refresh 5 minutes before expiration const refreshTime = (expiresIn - 300) * 1000;

this.refreshTimer = setTimeout(() => {
  this.refreshAccessToken();
}, refreshTime);

}

async refreshAccessToken() { try { const newTokens = await oauthClient.refresh(); this.storeTokens(newTokens); this.scheduleRefresh(newTokens.expires_in); } catch (error) { // Refresh failed, need re-authentication this.handleReauthentication(); } } }

Always use latest refresh token

• Previous refresh tokens expire 24 hours after new one issued

• Store refresh token immediately after refresh

• Never use old refresh tokens

Store tokens securely

• Encrypt in database

• Never commit to version control

• Use environment variables for development

Handle 401 responses automatically

def api_call_with_auto_refresh(api_function): try: return api_function() except Unauthorized401Error: # Attempt token refresh refresh_tokens() # Retry with new token return api_function()

API Rate Limiting

Implement exponential backoff for 429

def call_with_rate_limit_handling(api_function): max_retries = 5 base_delay = 1

for attempt in range(max_retries):
    try:
        return api_function()
    except RateLimitError as e:
        if attempt == max_retries - 1:
            raise

        delay = base_delay * (2 ** attempt)  # 1s, 2s, 4s, 8s, 16s
        time.sleep(delay)
        continue

- Use batch operations to reduce call count

• 1 batch request vs 30 individual = 30x reduction

• Batch counts as single API call for rate limits

Monitor rate limit headers (if provided)

• Some endpoints return rate limit info in headers

• Track usage to stay within limits

Multi-currency Considerations

CurrencyRef required when multicurrency enabled

{ "Invoice": { "CurrencyRef": { "value": "USD", "name": "United States Dollar" } } }

Exchange rate handling

• API automatically applies exchange rates

• ExchangeRate field shows conversion rate used

• Home currency amounts calculated automatically

Locale-specific required fields

• France: DocNumber required if custom transaction numbers enabled

• UK: Different tax handling (VAT)

• Check locale-specific documentation

Testing & Development

Use sandbox companies (free with developer account)

• Create at developer.intuit.com

• Separate from production data

• Full API feature parity

Test OAuth flow end-to-end

• Authorization URL → code exchange → token refresh

• Test token expiration handling

• Verify refresh token rotation

Validate webhook endpoint

• Test with sample payloads

• Ensure < 1 second response time

• Handle webhook signature verification

Handle all fault types in production

• ValidationFault, SystemFault, AuthenticationFault, AuthorizationFault

• Log error details (code, message, element)

• Implement appropriate retry logic

Monitor API calls and errors

• Track success/failure rates

• Alert on elevated error rates

• Log intuit_tid for support requests

Common Workflows

Workflow 1: Create and Send Invoice

Scenario: Create an invoice for a customer and send via email.

Steps:

• Query or create customer

Check if customer exists

customers = query_customers_by_email(realm_id, "customer@example.com", access_token)

if not customers: # Create new customer customer = create_customer(realm_id, { "DisplayName": "Acme Corp", "PrimaryEmailAddr": {"Address": "customer@example.com"}, "BillAddr": { "Line1": "123 Main St", "City": "San Francisco", "CountrySubDivisionCode": "CA", "PostalCode": "94105" } }, access_token) else: customer = customers[0]

customer_id = customer['Id']

• Query items for line items

Get service item

query = "SELECT * FROM Item WHERE Type = 'Service' AND Name = 'Consulting'" items = query_entity(realm_id, query, access_token) service_item = items[0]

• Create invoice with line items

invoice_data = { "TxnDate": "2024-12-09", "DueDate": "2024-12-23", "CustomerRef": {"value": customer_id}, "BillEmail": {"Address": "customer@example.com"}, "EmailStatus": "NeedToSend", # Mark for email sending "Line": [ { "Amount": 1500.00, "DetailType": "SalesItemLineDetail", "SalesItemLineDetail": { "ItemRef": {"value": service_item['Id']}, "Qty": 10, "UnitPrice": 150.00, "TaxCodeRef": {"value": "NON"} # Non-taxable }, "Description": "Consulting services - December 2024" }, { "Amount": 1500.00, "DetailType": "SubTotalLineDetail", "SubTotalLineDetail": {} } ] }

invoice = create_invoice(realm_id, invoice_data, access_token) print(f"Invoice {invoice['DocNumber']} created: ${invoice['TotalAmt']}")

• Send invoice email (automatic if EmailStatus = "NeedToSend")

QuickBooks automatically sends email when EmailStatus is NeedToSend

Alternatively, use send endpoint:

send_url = f"{base_url}/company/{realm_id}/invoice/{invoice['Id']}/send" params = {"sendTo": "customer@example.com"}

response = requests.post(send_url, params=params, headers=headers) if response.status_code == 200: print(f"Invoice sent to {customer['PrimaryEmailAddr']['Address']}")

• Handle response and linked transactions

Check invoice status

print(f"Invoice ID: {invoice['Id']}") print(f"Balance: ${invoice['Balance']}") print(f"Email Status: {invoice['EmailStatus']}")

Track linked transactions

if 'LinkedTxn' in invoice: for linked in invoice['LinkedTxn']: print(f"Linked {linked['TxnType']}: {linked['TxnId']}")

Workflow 2: Record Payment Against Invoice

Scenario: Customer pays an invoice via check.

Steps:

• Query invoice by DocNumber

def find_invoice_by_number(realm_id, doc_number, access_token): query = f"SELECT * FROM Invoice WHERE DocNumber = '{doc_number}'" invoices = query_entity(realm_id, query, access_token)

if not invoices:
    raise ValueError(f"Invoice {doc_number} not found")

return invoices[0]

invoice = find_invoice_by_number(realm_id, "1045", access_token) customer_id = invoice['CustomerRef']['value'] balance = invoice['Balance']

• Create payment entity

Get payment method (Check)

payment_methods = query_entity(realm_id, "SELECT * FROM PaymentMethod WHERE Name = 'Check'", access_token) payment_method_id = payment_methods[0]['Id']

payment_data = { "TotalAmt": balance, # Pay full amount "CustomerRef": {"value": customer_id}, "PaymentMethodRef": {"value": payment_method_id}, "PaymentRefNum": "1234", # Check number "TxnDate": "2024-12-09", "Line": [ { "Amount": balance, "LinkedTxn": [ { "TxnId": invoice['Id'], "TxnType": "Invoice" } ] } ] }

payment = create_payment(realm_id, payment_data, access_token)

• Verify balance updates

Re-read invoice to see updated balance

updated_invoice = read_invoice(realm_id, invoice['Id'], access_token)

print(f"Original balance: ${balance}") print(f"Payment amount: ${payment['TotalAmt']}") print(f"New balance: ${updated_invoice['Balance']}") print(f"Unapplied payment amount: ${payment.get('UnappliedAmt', 0)}")

• Handle partial payments

def apply_partial_payment(realm_id, invoice_id, payment_amount, customer_id, access_token): payment_data = { "TotalAmt": payment_amount, # Less than invoice balance "CustomerRef": {"value": customer_id}, "Line": [ { "Amount": payment_amount, "LinkedTxn": [ { "TxnId": invoice_id, "TxnType": "Invoice" } ] } ] }

payment = create_payment(realm_id, payment_data, access_token)

# Check unapplied amount
if payment['UnappliedAmt'] > 0:
    print(f"Warning: ${payment['UnappliedAmt']} unapplied (overpayment or error)")

return payment

• Apply payment to multiple invoices

def pay_multiple_invoices(realm_id, invoice_ids, amounts, customer_id, total_paid, access_token): lines = []

for invoice_id, amount in zip(invoice_ids, amounts):
    lines.append({
        "Amount": amount,
        "LinkedTxn": [{
            "TxnId": invoice_id,
            "TxnType": "Invoice"
        }]
    })

payment_data = {
    "TotalAmt": total_paid,
    "CustomerRef": {"value": customer_id},
    "Line": lines
}

return create_payment(realm_id, payment_data, access_token)

Example: Pay two invoices with single check

payment = pay_multiple_invoices( realm_id, ["145", "146"], # Invoice IDs [100.00, 50.00], # Amounts applied to each customer_id, 150.00, # Total check amount access_token )

Workflow 3: Customer Management

Scenario: Complete customer lifecycle management.

1. Create customer with address

def create_customer_complete(realm_id, customer_info, access_token): customer_data = { "DisplayName": customer_info['display_name'], "GivenName": customer_info.get('first_name'), "FamilyName": customer_info.get('last_name'), "CompanyName": customer_info.get('company_name'), "PrimaryEmailAddr": { "Address": customer_info['email'] }, "PrimaryPhone": { "FreeFormNumber": customer_info.get('phone') }, "BillAddr": { "Line1": customer_info['address_line1'], "City": customer_info['city'], "CountrySubDivisionCode": customer_info['state'], "PostalCode": customer_info['zip'] }, "ShipAddr": { "Line1": customer_info.get('ship_line1', customer_info['address_line1']), "City": customer_info.get('ship_city', customer_info['city']), "CountrySubDivisionCode": customer_info.get('ship_state', customer_info['state']), "PostalCode": customer_info.get('ship_zip', customer_info['zip']) } }

return create_customer(realm_id, customer_data, access_token)

2. Sparse update to modify email

def update_customer_email(realm_id, customer_id, new_email, access_token): # Read current customer customer = read_customer(realm_id, customer_id, access_token)

# Sparse update - only email
update_data = {
    "Id": customer_id,
    "SyncToken": customer['SyncToken'],
    "sparse": True,
    "PrimaryEmailAddr": {
        "Address": new_email
    }
}

return update_customer(realm_id, update_data, access_token)

3. Query customer transactions

def get_customer_transactions(realm_id, customer_id, access_token): transactions = {}

# Query invoices
invoice_query = f"SELECT * FROM Invoice WHERE CustomerRef = '{customer_id}'"
transactions['invoices'] = query_entity(realm_id, invoice_query, access_token)

# Query payments
payment_query = f"SELECT * FROM Payment WHERE CustomerRef = '{customer_id}'"
transactions['payments'] = query_entity(realm_id, payment_query, access_token)

# Query estimates
estimate_query = f"SELECT * FROM Estimate WHERE CustomerRef = '{customer_id}'"
transactions['estimates'] = query_entity(realm_id, estimate_query, access_token)

# Calculate totals
total_invoiced = sum(inv['TotalAmt'] for inv in transactions['invoices'])
total_paid = sum(pmt['TotalAmt'] for pmt in transactions['payments'])

transactions['summary'] = {
    'total_invoiced': total_invoiced,
    'total_paid': total_paid,
    'balance': total_invoiced - total_paid
}

return transactions

4. Update AR account reference

Change default AR account for customer

def update_customer_ar_account(realm_id, customer_id, new_ar_account_id, access_token): customer = read_customer(realm_id, customer_id, access_token)

update_data = {
    "Id": customer_id,
    "SyncToken": customer['SyncToken'],
    "sparse": True,
    "ARAccountRef": {
        "value": new_ar_account_id
    }
}

return update_customer(realm_id, update_data, access_token)

Workflow 4: Batch Sync Operation

Scenario: Sync changed entities using CDC and batch updates.

1. Use CDC to get changed entities

from datetime import datetime, timedelta

def sync_changed_entities(realm_id, last_sync_time, access_token): # Get changes since last sync entity_types = ['Invoice', 'Customer', 'Payment', 'Item'] changes = get_changed_entities(realm_id, entity_types, last_sync_time, access_token)

return changes

2. Build batch operation with updates

def build_batch_updates(changes): batch_items = [] bid_counter = 0

# Process each entity type
for entity_type in ['Customer', 'Invoice', 'Payment']:
    entities = extract_entities_by_type(changes, entity_type)

    for entity in entities:
        if entity.get('status') == 'Deleted':
            # Skip deleted entities or handle separately
            continue

        # Example: Mark all invoices as reviewed
        if entity_type == 'Invoice':
            batch_items.append({
                "bId": f"invoice_{bid_counter}",
                "operation": "update",
                "Invoice": {
                    "Id": entity['Id'],
                    "SyncToken": entity['SyncToken'],
                    "sparse": True,
                    "PrivateNote": f"Synced at {datetime.now().isoformat()}"
                }
            })
            bid_counter += 1

return batch_items

3. Execute batch asynchronously

async def execute_batch_sync(realm_id, batch_items, access_token): # Split into batches of 30 (API limit) batch_size = 30 results = []

for i in range(0, len(batch_items), batch_size):
    batch_chunk = batch_items[i:i+batch_size]

    batch_request = {
        "BatchItemRequest": batch_chunk
    }

    url = f"https://sandbox-quickbooks.api.intuit.com/v3/company/{realm_id}/batch"

    response = await async_post(url, batch_request, {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json"
    })

    if response.status == 200:
        batch_response = response.json()
        results.extend(batch_response['BatchItemResponse'])
    else:
        print(f"Batch {i//batch_size + 1} failed: {response.text}")

return results

4. Process batch responses by batch ID

def process_batch_results(batch_results): success_count = 0 error_count = 0 errors = []

for result in batch_results:
    bid = result['bId']

    if 'Fault' in result:
        error_count += 1
        fault = result['Fault']
        errors.append({
            'batch_id': bid,
            'error_code': fault['Error'][0]['code'],
            'message': fault['Error'][0]['Message']
        })
        print(f"Error in {bid}: {fault['Error'][0]['Message']}")
    else:
        success_count += 1
        # Extract updated entity
        entity_type = list(result.keys())[0]
        if entity_type != 'bId':
            entity = result[entity_type]
            print(f"Success {bid}: {entity_type} {entity['Id']} updated")

summary = {
    'total': len(batch_results),
    'success': success_count,
    'errors': error_count,
    'error_details': errors
}

return summary

Complete workflow

async def sync_workflow(realm_id, last_sync_time, access_token): # 1. Get changes via CDC changes = sync_changed_entities(realm_id, last_sync_time, access_token)

# 2. Build batch updates
batch_items = build_batch_updates(changes)

if not batch_items:
    print("No changes to sync")
    return

# 3. Execute batch
results = await execute_batch_sync(realm_id, batch_items, access_token)

# 4. Process results
summary = process_batch_results(results)

print(f"Sync complete: {summary['success']}/{summary['total']} successful")
if summary['errors'] > 0:
    print(f"Errors encountered: {summary['errors']}")
    for error in summary['error_details']:
        print(f"  {error['batch_id']}: {error['message']}")

return summary

Code Examples

Example 1: OAuth2 Token Refresh (Node.js)

Complete token management with automatic refresh:

const OAuthClient = require('intuit-oauth');

class QuickBooksAuth { constructor(clientId, clientSecret, redirectUri, environment = 'sandbox') { this.oauthClient = new OAuthClient({ clientId: clientId, clientSecret: clientSecret, environment: environment, redirectUri: redirectUri });

this.accessToken = null;
this.refreshToken = null;
this.tokenExpiry = null;
this.refreshTimer = null;

}

// Store tokens after authorization async storeTokens(authResponse) { this.accessToken = authResponse.token.access_token; this.refreshToken = authResponse.token.refresh_token;

// Calculate expiry time
const expiresIn = authResponse.token.expires_in; // 3600 seconds
this.tokenExpiry = Date.now() + (expiresIn * 1000);

// Schedule automatic refresh (5 minutes before expiry)
this.scheduleRefresh(expiresIn - 300);

// Persist tokens to secure storage
await this.saveToDatabase({
  access_token: this.accessToken,
  refresh_token: this.refreshToken,
  expiry: this.tokenExpiry
});

}

// Schedule automatic token refresh scheduleRefresh(delaySeconds) { if (this.refreshTimer) { clearTimeout(this.refreshTimer); }

this.refreshTimer = setTimeout(async () => {
  try {
    await this.refreshAccessToken();
  } catch (error) {
    console.error('Scheduled token refresh failed:', error);
    // Notify admin that re-authentication needed
    this.notifyReauthenticationNeeded();
  }
}, delaySeconds * 1000);

}

// Refresh access token async refreshAccessToken() { try { // Set refresh token in client this.oauthClient.setToken({ refresh_token: this.refreshToken });

  // Refresh
  const authResponse = await this.oauthClient.refresh();

  console.log('Token refreshed successfully');

  // Store new tokens
  await this.storeTokens(authResponse);

  return authResponse;

} catch (error) {
  console.error('Token refresh failed:', error.originalMessage);

  // Check if refresh token is invalid
  if (error.error === 'invalid_grant') {
    console.error('Refresh token invalid - re-authentication required');
    this.accessToken = null;
    this.refreshToken = null;
    throw new Error('Re-authentication required');
  }

  throw error;
}

}

// Get valid access token (refresh if needed) async getAccessToken() { // Check if token is about to expire (within 5 minutes) const bufferTime = 5 * 60 * 1000; // 5 minutes

if (!this.accessToken || Date.now() >= (this.tokenExpiry - bufferTime)) {
  console.log('Token expired or expiring soon, refreshing...');
  await this.refreshAccessToken();
}

return this.accessToken;

}

// Make API call with automatic token refresh async apiCall(url, options = {}) { try { const token = await this.getAccessToken();

  const response = await fetch(url, {
    ...options,
    headers: {
      ...options.headers,
      'Authorization': `Bearer ${token}`,
      'Accept': 'application/json'
    }
  });

  // Handle 401 - token might have expired
  if (response.status === 401) {
    console.log('401 Unauthorized - refreshing token and retrying...');
    await this.refreshAccessToken();

    // Retry with new token
    const newToken = await this.getAccessToken();
    return fetch(url, {
      ...options,
      headers: {
        ...options.headers,
        'Authorization': `Bearer ${newToken}`,
        'Accept': 'application/json'
      }
    });
  }

  return response;

} catch (error) {
  console.error('API call failed:', error);
  throw error;
}

}

// Save tokens to database (implement based on your storage) async saveToDatabase(tokens) { // Example: Save to database // await db.tokens.update({ realmId }, tokens); }

// Notify admin about re-auth requirement notifyReauthenticationNeeded() { // Example: Send email or notification console.error('Re-authentication required for QuickBooks integration'); } }

// Usage const auth = new QuickBooksAuth( 'YOUR_CLIENT_ID', 'YOUR_CLIENT_SECRET', 'https://yourapp.com/callback', 'sandbox' );

// After OAuth authorization auth.storeTokens(authResponse);

// Make API calls - automatic token refresh const response = await auth.apiCall( 'https://sandbox-quickbooks.api.intuit.com/v3/company/123/customer/456', { method: 'GET' } );

Example 2: Create Invoice (Python)

Complete invoice creation with line items and tax:

import requests from datetime import datetime, timedelta

class QuickBooksInvoice: def init(self, realm_id, access_token, base_url='https://sandbox-quickbooks.api.intuit.com'): self.realm_id = realm_id self.access_token = access_token self.base_url = base_url

def create_invoice(self, customer_id, line_items, due_days=30, tax_code='TAX', memo=None):
    """
    Create an invoice with multiple line items

    Args:
        customer_id: QuickBooks customer ID
        line_items: List of dicts with 'item_id', 'quantity', 'unit_price', 'description'
        due_days: Days until due date
        tax_code: Tax code ('TAX' for taxable, 'NON' for non-taxable)
        memo: Customer memo

    Returns:
        Created invoice dict or None if error
    """
    # Calculate dates
    txn_date = datetime.now().strftime('%Y-%m-%d')
    due_date = (datetime.now() + timedelta(days=due_days)).strftime('%Y-%m-%d')

    # Build line items
    lines = []
    subtotal = 0

    for idx, item in enumerate(line_items, start=1):
        amount = item['quantity'] * item['unit_price']
        subtotal += amount

        lines.append({
            "LineNum": idx,
            "Amount": amount,
            "DetailType": "SalesItemLineDetail",
            "Description": item.get('description', ''),
            "SalesItemLineDetail": {
                "ItemRef": {
                    "value": item['item_id']
                },
                "Qty": item['quantity'],
                "UnitPrice": item['unit_price'],
                "TaxCodeRef": {
                    "value": tax_code
                }
            }
        })

    # Add subtotal line
    lines.append({
        "Amount": subtotal,
        "DetailType": "SubTotalLineDetail",
        "SubTotalLineDetail": {}
    })

    # Build invoice payload
    invoice_data = {
        "TxnDate": txn_date,
        "DueDate": due_date,
        "CustomerRef": {
            "value": customer_id
        },
        "Line": lines,
        "BillEmail": {},  # Will be populated from customer
        "EmailStatus": "NotSet"
    }

    # Add memo if provided
    if memo:
        invoice_data["CustomerMemo"] = {
            "value": memo
        }

    # Make API request
    url = f"{self.base_url}/v3/company/{self.realm_id}/invoice"

    headers = {
        "Authorization": f"Bearer {self.access_token}",
        "Accept": "application/json",
        "Content-Type": "application/json"
    }

    try:
        response = requests.post(url, json=invoice_data, headers=headers)
        response.raise_for_status()

        # Check for fault in response
        result = response.json()

        if 'Fault' in result:
            self._handle_fault(result['Fault'])
            return None

        invoice = result['Invoice']

        print(f"✓ Invoice {invoice['DocNumber']} created")
        print(f"  Customer: {invoice['CustomerRef']['value']}")
        print(f"  Total: ${invoice['TotalAmt']:.2f}")
        print(f"  Due: {invoice['DueDate']}")
        print(f"  Balance: ${invoice['Balance']:.2f}")

        return invoice

    except requests.exceptions.HTTPError as e:
        print(f"✗ HTTP Error: {e.response.status_code}")
        print(f"  Response: {e.response.text}")
        return None
    except Exception as e:
        print(f"✗ Error creating invoice: {str(e)}")
        return None

def send_invoice(self, invoice_id, email_address):
    """Send invoice via email"""
    url = f"{self.base_url}/v3/company/{self.realm_id}/invoice/{invoice_id}/send"

    headers = {
        "Authorization": f"Bearer {self.access_token}",
        "Accept": "application/json"
    }

    params = {"sendTo": email_address}

    try:
        response = requests.post(url, params=params, headers=headers)
        response.raise_for_status()

        result = response.json()

        if 'Fault' in result:
            self._handle_fault(result['Fault'])
            return False

        invoice = result['Invoice']
        print(f"✓ Invoice {invoice['DocNumber']} sent to {email_address}")
        print(f"  Email Status: {invoice['EmailStatus']}")

        return True

    except Exception as e:
        print(f"✗ Error sending invoice: {str(e)}")
        return False

def _handle_fault(self, fault):
    """Handle fault responses"""
    print(f"✗ Fault Type: {fault['type']}")
    for error in fault['Error']:
        print(f"  Error {error['code']}: {error['Message']}")
        if 'element' in error:
            print(f"  Element: {error['element']}")

Usage example

invoice_manager = QuickBooksInvoice(realm_id='123456789', access_token='your_token')

Create invoice with multiple items

invoice = invoice_manager.create_invoice( customer_id='42', line_items=[ { 'item_id': '1', 'quantity': 10, 'unit_price': 150.00, 'description': 'Consulting services - December 2024' }, { 'item_id': '5', 'quantity': 1, 'unit_price': 500.00, 'description': 'Project management - December 2024' } ], due_days=30, tax_code='TAX', # or 'NON' for non-taxable memo='Thank you for your business!' )

if invoice: # Send invoice via email invoice_manager.send_invoice(invoice['Id'], 'customer@example.com')

Example 3: Sparse Update Customer (Node.js)

Demonstrate sparse update pattern with error handling:

const axios = require('axios');

class QuickBooksCustomer { constructor(realmId, accessToken, baseUrl = 'https://sandbox-quickbooks.api.intuit.com') { this.realmId = realmId; this.accessToken = accessToken; this.baseUrl = baseUrl; }

async readCustomer(customerId) { const url = ${this.baseUrl}/v3/company/${this.realmId}/customer/${customerId};

try {
  const response = await axios.get(url, {
    headers: {
      'Authorization': `Bearer ${this.accessToken}`,
      'Accept': 'application/json'
    }
  });

  return response.data.Customer;
} catch (error) {
  console.error('Read customer failed:', error.response?.data || error.message);
  throw error;
}

}

async sparseUpdate(customerId, updates) { // First, read customer to get current SyncToken const customer = await this.readCustomer(customerId);

// Build sparse update payload
const updateData = {
  Id: customerId,
  SyncToken: customer.SyncToken,
  sparse: true,
  ...updates
};

const url = `${this.baseUrl}/v3/company/${this.realmId}/customer`;

try {
  const response = await axios.post(url, updateData, {
    headers: {
      'Authorization': `Bearer ${this.accessToken}`,
      'Accept': 'application/json',
      'Content-Type': 'application/json'
    }
  });

  // Check for fault
  if (response.data.Fault) {
    this.handleFault(response.data.Fault);
    return null;
  }

  const updatedCustomer = response.data.Customer;

  console.log(`✓ Customer ${updatedCustomer.DisplayName} updated`);
  console.log(`  New SyncToken: ${updatedCustomer.SyncToken}`);

  return updatedCustomer;

} catch (error) {
  if (error.response?.status === 400) {
    const fault = error.response.data.Fault;

    // Handle SyncToken mismatch
    if (fault.Error[0].code === '3200') {
      console.log('SyncToken mismatch - retrying with fresh token...');
      // Recursive retry with new token
      return this.sparseUpdate(customerId, updates);
    }
  }

  console.error('Update failed:', error.response?.data || error.message);
  throw error;
}

}

// Example: Update email async updateEmail(customerId, newEmail) { return this.sparseUpdate(customerId, { PrimaryEmailAddr: { Address: newEmail } }); }

// Example: Update phone async updatePhone(customerId, newPhone) { return this.sparseUpdate(customerId, { PrimaryPhone: { FreeFormNumber: newPhone } }); }

// Example: Update billing address async updateBillingAddress(customerId, address) { return this.sparseUpdate(customerId, { BillAddr: { Line1: address.line1, City: address.city, CountrySubDivisionCode: address.state, PostalCode: address.zip } }); }

// Example: Deactivate customer async deactivateCustomer(customerId) { return this.sparseUpdate(customerId, { Active: false }); }

// Example: Update multiple fields at once async updateMultipleFields(customerId, updates) { const sparseUpdates = {};

if (updates.email) {
  sparseUpdates.PrimaryEmailAddr = { Address: updates.email };
}

if (updates.phone) {
  sparseUpdates.PrimaryPhone = { FreeFormNumber: updates.phone };
}

if (updates.displayName) {
  sparseUpdates.DisplayName = updates.displayName;
}

if (updates.notes) {
  sparseUpdates.Notes = updates.notes;
}

return this.sparseUpdate(customerId, sparseUpdates);

}

handleFault(fault) { console.error(✗ Fault Type: ${fault.type}); fault.Error.forEach(error => { console.error( Error ${error.code}: ${error.Message}); if (error.element) { console.error( Element: ${error.element}); } }); } }

// Usage const customerManager = new QuickBooksCustomer('123456789', 'your_access_token');

// Update email await customerManager.updateEmail('42', 'newemail@example.com');

// Update phone await customerManager.updatePhone('42', '(555) 987-6543');

// Update address await customerManager.updateBillingAddress('42', { line1: '456 New Street', city: 'San Francisco', state: 'CA', zip: '94105' });

// Update multiple fields await customerManager.updateMultipleFields('42', { email: 'updated@example.com', phone: '(555) 111-2222', displayName: 'Updated Customer Name', notes: 'VIP customer - priority support' });

// Deactivate customer await customerManager.deactivateCustomer('42');

Example 4: Query with Filters (Python)

Complex query with date range and sorting:

import requests from urllib.parse import quote from datetime import datetime, timedelta

class QuickBooksQuery: def init(self, realm_id, access_token, base_url='https://sandbox-quickbooks.api.intuit.com'): self.realm_id = realm_id self.access_token = access_token self.base_url = base_url

def query(self, sql_query):
    """Execute SQL-like query"""
    encoded_query = quote(sql_query)
    url = f"{self.base_url}/v3/company/{self.realm_id}/query?query={encoded_query}"

    headers = {
        "Authorization": f"Bearer {self.access_token}",
        "Accept": "application/json"
    }

    try:
        response = requests.get(url, headers=headers)
        response.raise_for_status()

        result = response.json()

        if 'Fault' in result:
            print(f"Query error: {result['Fault']}")
            return []

        query_response = result.get('QueryResponse', {})

        # Extract entities (keys vary by entity type)
        for key, value in query_response.items():
            if key not in ['startPosition', 'maxResults', 'totalCount']:
                return value if isinstance(value, list) else []

        return []

    except Exception as e:
        print(f"Query failed: {str(e)}")
        return []

def query_invoices_by_date_range(self, start_date, end_date, customer_id=None):
    """Query invoices within date range, optionally filtered by customer"""
    query = f"SELECT * FROM Invoice WHERE TxnDate >= '{start_date}' AND TxnDate <= '{end_date}'"

    if customer_id:
        query += f" AND CustomerRef = '{customer_id}'"

    query += " ORDERBY TxnDate DESC"

    invoices = self.query(query)

    print(f"Found {len(invoices)} invoices between {start_date} and {end_date}")

    # Calculate totals
    total_amount = sum(inv['TotalAmt'] for inv in invoices)
    total_balance = sum(inv['Balance'] for inv in invoices)

    print(f"Total invoiced: ${total_amount:.2f}")
    print(f"Outstanding balance: ${total_balance:.2f}")

    return invoices

def query_overdue_invoices(self, as_of_date=None):
    """Query invoices past due date"""
    if not as_of_date:
        as_of_date = datetime.now().strftime('%Y-%m-%d')

    query = f"SELECT * FROM Invoice WHERE Balance > '0' AND DueDate < '{as_of_date}' ORDERBY DueDate"

    invoices = self.query(query)

    print(f"Found {len(invoices)} overdue invoices as of {as_of_date}")

    # Group by customer
    by_customer = {}
    for inv in invoices:
        customer_id = inv['CustomerRef']['value']
        if customer_id not in by_customer:
            by_customer[customer_id] = {
                'customer_name': inv['CustomerRef'].get('name', 'Unknown'),
                'invoices': [],
                'total_overdue': 0
            }

        by_customer[customer_id]['invoices'].append(inv)
        by_customer[customer_id]['total_overdue'] += inv['Balance']

    # Print summary
    for customer_id, data in by_customer.items():
        print(f"\nCustomer: {data['customer_name']}")
        print(f"  Overdue invoices: {len(data['invoices'])}")
        print(f"  Total overdue: ${data['total_overdue']:.2f}")

    return invoices

def query_customers_by_balance(self, min_balance=0):
    """Query customers with balance greater than minimum"""
    query = f"SELECT * FROM Customer WHERE Balance > '{min_balance}' ORDERBY Balance DESC"

    customers = self.query(query)

    print(f"Found {len(customers)} customers with balance > ${min_balance}")

    total_ar = sum(cust['Balance'] for cust in customers)
    print(f"Total accounts receivable: ${total_ar:.2f}")

    return customers

def query_items_by_type(self, item_type='Service'):
    """Query items by type (Service, Inventory, NonInventory, Category)"""
    query = f"SELECT * FROM Item WHERE Type = '{item_type}' AND Active = true ORDERBY Name"

    items = self.query(query)

    print(f"Found {len(items)} active {item_type} items")

    return items

def search_customers_by_name(self, search_term):
    """Search customers by display name"""
    query = f"SELECT * FROM Customer WHERE DisplayName LIKE '%{search_term}%' ORDERBY DisplayName"

    customers = self.query(query)

    print(f"Found {len(customers)} customers matching '{search_term}'")

    for cust in customers:
        print(f"  {cust['DisplayName']} - Balance: ${cust['Balance']:.2f}")

    return customers

def query_recent_payments(self, days=30):
    """Query payments from last N days"""
    start_date = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')

    query = f"SELECT * FROM Payment WHERE TxnDate >= '{start_date}' ORDERBY TxnDate DESC"

    payments = self.query(query)

    print(f"Found {len(payments)} payments in last {days} days")

    total_received = sum(pmt['TotalAmt'] for pmt in payments)
    print(f"Total payments received: ${total_received:.2f}")

    return payments

Usage

query_service = QuickBooksQuery(realm_id='123456789', access_token='your_token')

Query invoices for date range

invoices = query_service.query_invoices_by_date_range( start_date='2024-01-01', end_date='2024-12-31' )

Query invoices for specific customer

customer_invoices = query_service.query_invoices_by_date_range( start_date='2024-01-01', end_date='2024-12-31', customer_id='42' )

Find overdue invoices

overdue = query_service.query_overdue_invoices()

Find customers with high balances

high_balance_customers = query_service.query_customers_by_balance(min_balance=1000.00)

Search for customer

customers = query_service.search_customers_by_name('Acme')

Get recent payments

recent_payments = query_service.query_recent_payments(days=30)

Example 5: Batch Operations (Node.js)

Batch create/update multiple entities:

const axios = require('axios');

class QuickBooksBatch { constructor(realmId, accessToken, baseUrl = 'https://sandbox-quickbooks.api.intuit.com') { this.realmId = realmId; this.accessToken = accessToken; this.baseUrl = baseUrl; }

async executeBatch(batchItems) { const url = ${this.baseUrl}/v3/company/${this.realmId}/batch;

try {
  const response = await axios.post(url, {
    BatchItemRequest: batchItems
  }, {
    headers: {
      'Authorization': `Bearer ${this.accessToken}`,
      'Content-Type': 'application/json',
      'Accept': 'application/json'
    }
  });

  const results = response.data.BatchItemResponse;

  // Process results
  const summary = {
    total: results.length,
    success: 0,
    errors: 0,
    results: []
  };

  results.forEach(result => {
    if (result.Fault) {
      summary.errors++;
      console.error(`✗ Error for ${result.bId}:`);
      result.Fault.Error.forEach(err => {
        console.error(`  ${err.code}: ${err.Message}`);
      });
      summary.results.push({
        bId: result.bId,
        status: 'error',
        error: result.Fault
      });
    } else {
      summary.success++;
      // Extract entity from result
      const entityType = Object.keys(result).find(k => k !== 'bId');
      const entity = result[entityType];

      console.log(`✓ Success for ${result.bId}: ${entityType} ${entity.Id}`);
      summary.results.push({
        bId: result.bId,
        status: 'success',
        entityType: entityType,
        entity: entity
      });
    }
  });

  console.log(`\nBatch complete: ${summary.success}/${summary.total} successful`);

  return summary;

} catch (error) {
  console.error('Batch operation failed:', error.response?.data || error.message);
  throw error;
}

}

// Batch create customers async batchCreateCustomers(customers) { const batchItems = customers.map((customer, index) => ({ bId: customer_create_${index}, operation: 'create', Customer: { DisplayName: customer.displayName, PrimaryEmailAddr: { Address: customer.email }, PrimaryPhone: { FreeFormNumber: customer.phone }, BillAddr: { Line1: customer.address, City: customer.city, CountrySubDivisionCode: customer.state, PostalCode: customer.zip } } }));

return this.executeBatch(batchItems);

}

// Batch update invoices async batchUpdateInvoices(updates) { const batchItems = updates.map((update, index) => ({ bId: invoice_update_${index}, operation: 'update', Invoice: { Id: update.id, SyncToken: update.syncToken, sparse: true, ...update.changes } }));

return this.executeBatch(batchItems);

}

// Batch query multiple entities async batchQuery(queries) { const batchItems = queries.map((query, index) => ({ bId: query_${index}, operation: 'query', Query: query.sql }));

const results = await this.executeBatch(batchItems);

// Extract query results
const queryResults = {};
results.results.forEach(result => {
  if (result.status === 'success' && result.entity.QueryResponse) {
    queryResults[result.bId] = result.entity.QueryResponse;
  }
});

return queryResults;

}

// Mixed batch operations async mixedBatch(operations) { const batchItems = operations.map((op, index) => { const item = { bId: op_${index}_${op.type}, operation: op.operation };

  // Add entity or query data
  if (op.operation === 'query') {
    item.Query = op.data;
  } else {
    item[op.entityType] = op.data;
  }

  return item;
});

return this.executeBatch(batchItems);

} }

// Usage Examples

const batchService = new QuickBooksBatch('123456789', 'your_access_token');

// Example 1: Batch create customers const newCustomers = [ { displayName: 'Acme Corp', email: 'contact@acme.com', phone: '(555) 111-1111', address: '123 Main St', city: 'San Francisco', state: 'CA', zip: '94105' }, { displayName: 'TechStart Inc', email: 'hello@techstart.com', phone: '(555) 222-2222', address: '456 Market St', city: 'San Francisco', state: 'CA', zip: '94103' } ];

const createResults = await batchService.batchCreateCustomers(newCustomers);

// Example 2: Batch update invoices (mark as sent) const invoiceUpdates = [ { id: '145', syncToken: '0', changes: { EmailStatus: 'NeedToSend' } }, { id: '146', syncToken: '0', changes: { EmailStatus: 'NeedToSend' } }, { id: '147', syncToken: '0', changes: { EmailStatus: 'NeedToSend' } } ];

const updateResults = await batchService.batchUpdateInvoices(invoiceUpdates);

// Example 3: Batch queries const queries = [ { sql: 'SELECT * FROM Customer WHERE Active = true MAXRESULTS 10' }, { sql: 'SELECT * FROM Invoice WHERE Balance > 0 MAXRESULTS 10' }, { sql: 'SELECT * FROM Payment MAXRESULTS 10' } ];

const queryResults = await batchService.batchQuery(queries);

// Example 4: Mixed batch operations const mixedOps = [ { type: 'create_customer', operation: 'create', entityType: 'Customer', data: { DisplayName: 'New Customer', PrimaryEmailAddr: { Address: 'new@example.com' } } }, { type: 'update_invoice', operation: 'update', entityType: 'Invoice', data: { Id: '145', SyncToken: '1', sparse: true, CustomerMemo: { value: 'Thank you!' } } }, { type: 'query_items', operation: 'query', data: 'SELECT * FROM Item WHERE Type = 'Service' MAXRESULTS 5' } ];

const mixedResults = await batchService.mixedBatch(mixedOps);

console.log(Mixed batch: ${mixedResults.success}/${mixedResults.total} successful);

Example 6: Payment Application (Python)

Apply payment to multiple invoices:

import requests

class QuickBooksPayment: def init(self, realm_id, access_token, base_url='https://sandbox-quickbooks.api.intuit.com'): self.realm_id = realm_id self.access_token = access_token self.base_url = base_url

def create_payment(self, customer_id, total_amount, payment_method_id,
                   payment_ref_num, txn_date, invoice_applications):
    """
    Create payment and apply to one or more invoices

    Args:
        customer_id: QuickBooks customer ID
        total_amount: Total payment amount
        payment_method_id: Payment method ID
        payment_ref_num: Check number or transaction reference
        txn_date: Payment date (YYYY-MM-DD)
        invoice_applications: List of {'invoice_id': str, 'amount': float}

    Returns:
        Created payment dict or None if error
    """
    # Build line items for invoice applications
    lines = []
    total_applied = 0

    for application in invoice_applications:
        lines.append({
            "Amount": application['amount'],
            "LinkedTxn": [
                {
                    "TxnId": application['invoice_id'],
                    "TxnType": "Invoice"
                }
            ]
        })
        total_applied += application['amount']

    # Check for unapplied amount
    unapplied = total_amount - total_applied
    if unapplied < 0:
        print(f"Warning: Applied amount (${total_applied}) exceeds payment (${total_amount})")
        return None

    # Build payment payload
    payment_data = {
        "TotalAmt": total_amount,
        "CustomerRef": {
            "value": customer_id
        },
        "TxnDate": txn_date,
        "PaymentMethodRef": {
            "value": payment_method_id
        },
        "PaymentRefNum": payment_ref_num,
        "Line": lines
    }

    # Make API request
    url = f"{self.base_url}/v3/company/{self.realm_id}/payment"

    headers = {
        "Authorization": f"Bearer {self.access_token}",
        "Accept": "application/json",
        "Content-Type": "application/json"
    }

    try:
        response = requests.post(url, json=payment_data, headers=headers)
        response.raise_for_status()

        result = response.json()

        if 'Fault' in result:
            self._handle_fault(result['Fault'])
            return None

        payment = result['Payment']

        print(f"✓ Payment created: ID {payment['Id']}")
        print(f"  Customer: {payment['CustomerRef']['value']}")
        print(f"  Total Amount: ${payment['TotalAmt']:.2f}")
        print(f"  Applied Amount: ${total_applied:.2f}")
        print(f"  Unapplied Amount: ${payment.get('UnappliedAmt', 0):.2f}")
        print(f"  Reference: {payment.get('PaymentRefNum', 'N/A')}")

        # Show invoice applications
        for line in payment['Line']:
            if 'LinkedTxn' in line:
                for linked in line['LinkedTxn']:
                    print(f"  Applied ${line['Amount']:.2f} to {linked['TxnType']} {linked['TxnId']}")

        return payment

    except requests.exceptions.HTTPError as e:
        print(f"✗ HTTP Error: {e.response.status_code}")
        print(f"  Response: {e.response.text}")
        return None
    except Exception as e:
        print(f"✗ Error creating payment: {str(e)}")
        return None

def apply_payment_to_invoices(self, customer_id, check_number, check_amount,
                               check_date, invoices):
    """
    Apply a single check payment to multiple invoices

    Args:
        customer_id: Customer ID
        check_number: Check number
        check_amount: Total check amount
        check_date: Check date
        invoices: List of {'id': str, 'amount_to_apply': float, 'balance': float}

    Returns:
        Payment dict or None
    """
    # Get check payment method ID
    payment_methods = self.query_payment_methods()
    check_method = next((pm for pm in payment_methods if pm['Name'].lower() == 'check'), None)

    if not check_method:
        print("Check payment method not found")
        return None

    # Build invoice applications
    applications = []
    total_to_apply = 0

    for invoice in invoices:
        amount = min(invoice['amount_to_apply'], invoice['balance'])
        applications.append({
            'invoice_id': invoice['id'],
            'amount': amount
        })
        total_to_apply += amount

        print(f"Will apply ${amount:.2f} to Invoice {invoice['id']}")

    # Check if payment covers all applications
    if total_to_apply > check_amount:
        print(f"Warning: Total applications (${total_to_apply}) exceeds check amount (${check_amount})")
        return None

    # Create payment
    payment = self.create_payment(
        customer_id=customer_id,
        total_amount=check_amount,
        payment_method_id=check_method['Id'],
        payment_ref_num=check_number,
        txn_date=check_date,
        invoice_applications=applications
    )

    return payment

def apply_partial_payment(self, customer_id, payment_amount, payment_method_id,
                          invoice_id, partial_amount):
    """Apply partial payment to invoice"""
    if partial_amount > payment_amount:
        print("Partial amount cannot exceed total payment")
        return None

    applications = [{
        'invoice_id': invoice_id,
        'amount': partial_amount
    }]

    payment = self.create_payment(
        customer_id=customer_id,
        total_amount=payment_amount,
        payment_method_id=payment_method_id,
        payment_ref_num='',
        txn_date=datetime.now().strftime('%Y-%m-%d'),
        invoice_applications=applications
    )

    if payment and payment.get('UnappliedAmt', 0) > 0:
        print(f"\nNote: ${payment['UnappliedAmt']:.2f} remains unapplied")
        print("This amount can be applied to future invoices or refunded")

    return payment

def query_payment_methods(self):
    """Get available payment methods"""
    from urllib.parse import quote

    query = "SELECT * FROM PaymentMethod"
    encoded_query = quote(query)

    url = f"{self.base_url}/v3/company/{self.realm_id}/query?query={encoded_query}"

    headers = {
        "Authorization": f"Bearer {self.access_token}",
        "Accept": "application/json"
    }

    response = requests.get(url, headers=headers)
    result = response.json()

    return result.get('QueryResponse', {}).get('PaymentMethod', [])

def _handle_fault(self, fault):
    """Handle fault responses"""
    print(f"✗ Fault Type: {fault['type']}")
    for error in fault['Error']:
        print(f"  Error {error['code']}: {error['Message']}")
        if 'element' in error:
            print(f"  Element: {error['element']}")

Usage Examples

payment_service = QuickBooksPayment(realm_id='123456789', access_token='your_token')

Example 1: Apply single check to multiple invoices

payment = payment_service.apply_payment_to_invoices( customer_id='42', check_number='1234', check_amount=1500.00, check_date='2024-12-09', invoices=[ {'id': '145', 'amount_to_apply': 1000.00, 'balance': 1000.00}, {'id': '146', 'amount_to_apply': 500.00, 'balance': 750.00} ] )

Example 2: Partial payment on single invoice

partial_payment = payment_service.apply_partial_payment( customer_id='42', payment_amount=500.00, payment_method_id='1', # Cash invoice_id='147', partial_amount=500.00 # Invoice balance is $1000, paying $500 )

Example 3: Payment with unapplied amount (credit for future invoices)

credit_payment = payment_service.create_payment( customer_id='42', total_amount=2000.00, # Customer pays $2000 payment_method_id='1', payment_ref_num='', txn_date='2024-12-09', invoice_applications=[ {'invoice_id': '145', 'amount': 1000.00} # Only $1000 applied ] # $1000 remains unapplied as credit )

API Reference Quick Links

Context7 Library: Use Context7 MCP to fetch latest documentation:

• Library ID: /websites/developer_intuit_app_developer_qbo

• Use for up-to-date code examples and API changes

Official Resources:

QuickBooks API Explorer: https://developer.intuit.com/app/developer/qbo/docs/api/accounting/all-entities/account

• Interactive API reference with sandbox testing

• Entity-specific documentation and sample requests

Developer Dashboard: https://developer.intuit.com/app/developer/myapps

• Manage apps, keys, webhooks

• Create sandbox companies

SDKs:

• Node.js: https://github.com/intuit/intuit-oauth (OAuth) + axios for API calls

• Python: https://github.com/intuit/intuit-oauth-python (OAuth) + requests

• Java: https://github.com/intuit/QuickBooks-V3-Java-SDK

• PHP: https://github.com/intuit/QuickBooks-V3-PHP-SDK

• C#/.NET: https://github.com/intuit/QuickBooks-V3-DotNET-SDK

Common Endpoints:

• Base URL (Sandbox): https://sandbox-quickbooks.api.intuit.com/v3/company/{realmId}

• Base URL (Production): https://quickbooks.api.intuit.com/v3/company/{realmId}

• Token endpoint: https://oauth.platform.intuit.com/oauth2/v1/tokens/bearer

• OAuth authorization: https://appcenter.intuit.com/connect/oauth2

Troubleshooting Common Issues

SyncToken Mismatch (Error 3200)

Symptom: "stale object error" when updating entities

Cause: SyncToken in request doesn't match current version (concurrent modification)

Solution:

def safe_update_with_retry(entity_id, updates, max_attempts=3): for attempt in range(max_attempts): try: # Read latest version entity = read_entity(entity_id)

        # Apply changes
        entity.update(updates)
        entity['sparse'] = True

        # Attempt update
        return update_entity(entity)

    except SyncTokenError as e:
        if attempt == max_attempts - 1:
            raise
        print(f"SyncToken mismatch, retrying... (attempt {attempt + 1})")
        continue

Required Field Missing (Error 6000)

Symptom: "business validation error" or "required field missing"

Cause: Missing required fields like TotalAmt, CustomerRef, or entity-specific requirements

Solution:

• Check API documentation for entity-specific required fields

• For Payment: TotalAmt and CustomerRef are required

• For Invoice: CustomerRef and Line array are required

• Validate data locally before API call

Common Required Fields:

• Customer: DisplayName (must be unique)

• Invoice: CustomerRef, Line (at least one)

• Payment: TotalAmt, CustomerRef

• Item: Name, Type, IncomeAccountRef (for Service)

OAuth Token Expiration (401 Unauthorized)

Symptom: "invalid_token" or "token_expired" errors

Cause: Access token expired (after 3600 seconds)

Solution:

async function apiCallWithAutoRefresh(apiFunction) { try { return await apiFunction(); } catch (error) { if (error.response?.status === 401) { // Token expired, refresh await refreshAccessToken(); // Retry with new token return await apiFunction(); } throw error; } }

Prevention:

• Implement proactive token refresh (every 50 minutes)

• Store token expiry time and check before requests

• Handle 401 responses automatically

Invalid Reference (Error 3100)

Symptom: "object not found" when referencing CustomerRef, ItemRef, etc.

Cause: Referenced entity doesn't exist or was deleted

Solution:

def validate_reference(entity_type, entity_id): """Verify entity exists before creating reference""" try: entity = read_entity(entity_type, entity_id) return True except NotFoundError: print(f"{entity_type} {entity_id} not found") return False

Before creating invoice

if validate_reference('Customer', customer_id): if validate_reference('Item', item_id): create_invoice(customer_id, item_id)

Rate Limiting (429 Too Many Requests)

Symptom: "throttle_limit_exceeded" or 429 status code

Cause: Exceeded API rate limits

Solution - Exponential backoff with jitter:

import time import random

def api_call_with_backoff(api_function, max_retries=5): for attempt in range(max_retries): try: return api_function() except RateLimitError: if attempt == max_retries - 1: raise

        # Exponential backoff with jitter
        delay = (2 ** attempt) + random.uniform(0, 1)
        print(f"Rate limited, waiting {delay:.1f}s...")
        time.sleep(delay)

Prevention:

• Use batch operations to reduce call count

• Implement request queuing with rate limiting

• Cache frequently accessed reference data

Batch Operation Failures

Symptom: Some operations in batch fail while others succeed

Cause: Each batch operation is independent; one failure doesn't affect others

Solution:

function processBatchResults(results) { const failed = results.filter(r => r.Fault); const succeeded = results.filter(r => !r.Fault);

console.log(Batch: ${succeeded.length} success, ${failed.length} failed);

// Retry failed operations individually for (const failure of failed) { console.log(Retrying ${failure.bId}...); // Implement individual retry logic }

return { succeeded, failed }; }

Multi-currency Validation Errors

Symptom: "currency not enabled" or "exchange rate required"

Cause: Multi-currency features not enabled or missing CurrencyRef

Solution:

{ "Invoice": { "CurrencyRef": { "value": "USD", "name": "United States Dollar" }, "ExchangeRate": 1.0 } }

Check:

• Verify multi-currency enabled in QuickBooks company preferences

• Always include CurrencyRef when multi-currency is enabled

• For foreign currency, API calculates exchange rate automatically

Webhook Not Receiving Notifications

Symptom: Webhook endpoint configured but not receiving POST requests

Cause: Endpoint issues, SSL problems, or slow response time

Solution:

• Verify endpoint is publicly accessible (not localhost)

• Use HTTPS (required for webhooks)

• Respond within 1 second (return 200 OK immediately, process async)

• Test with sample payload: curl -X POST https://yourapp.com/webhooks/quickbooks
  -H "Content-Type: application/json"
  -d '{"eventNotifications":[]}'

• Check webhook logs in developer dashboard

Deleted Entities in CDC Response

Symptom: Entities with status="Deleted" only contain ID

Cause: CDC returns minimal data for deleted entities

Solution:

def process_cdc_changes(changes): for entity in changes: if entity.get('status') == 'Deleted': # Only ID available handle_deletion(entity['Id']) else: # Full entity data available process_entity_update(entity)

This skill provides comprehensive guidance for QuickBooks Online API integration. For the most current API documentation and changes, use Context7 with library ID /websites/developer_intuit_app_developer_qbo .