# Quickbooks Online API

> Expert guide for QuickBooks Online API integration covering authentication, CRUD operations, batch processing, and best practices for invoicing, payments, and customer management.

- Skill: `freightcognition/quickbooks-online-api` (Agent Skill)
- Install (CLI): `npx skillmds@latest add freightcognition/quickbooks-online-api`
- Raw SKILL.md: https://api.skillmd.com/api/skills/freightcognition/quickbooks-online-api/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Integrations & APIs
- Author: freightcognition (https://skillmd.com/u/freightcognition)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/freightcognition/quickbooks-online-api

---


# 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:

1. **Register your app** at developer.intuit.com to get Client ID and Client Secret
2. **Direct users to authorization URL** where they grant access to their QuickBooks company
3. **Exchange authorization code for tokens** (access token + refresh token)
4. **Use access token** in API requests (Authorization: Bearer header)
5. **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**:
```javascript
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**:
```python
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**:
```json
{
  "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):
```json
{
  "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**:
```python
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**:
```javascript
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**:
```python
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**:
```python
# 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**:
```javascript
// 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`

```json
{
  "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**:
```sql
SELECT * FROM Customer WHERE DisplayName LIKE 'Acme%'
```

**Query invoices by date range**:
```sql
SELECT * FROM Invoice WHERE TxnDate >= '2024-01-01' AND TxnDate <= '2024-12-31'
```

**Query with ordering**:
```sql
SELECT * FROM Customer WHERE Active = true ORDERBY DisplayName
```

**Pagination**:
```sql
SELECT * FROM Invoice STARTPOSITION 1 MAXRESULTS 100
```

**Python Example - Query with Filters**:
```python
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

```python
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**:
```json
{
  "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**:
```json
{
  "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

```javascript
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:

```json
{
  "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**:

1. **ValidationFault**: Invalid request data or business rule violation
   - Fix: Correct request payload, check required fields

2. **SystemFault**: Server-side error
   - Fix: Retry request, contact support if persists

3. **AuthenticationFault**: Invalid credentials
   - Fix: Refresh access token, re-authenticate

4. **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**:
```python
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

1. **Check response body even with 200**: Fault elements can appear in successful responses
2. **Log intuit_tid**: Include in support requests for faster resolution
3. **Validate SyncToken**: Always use latest version from read operations
4. **Test in sandbox first**: Use sandbox companies for development
5. **Implement retry logic**: Exponential backoff for 500/503 errors
6. **Parse error details**: Check `error.code`, `element`, `message` fields

### Retry Pattern with Exponential Backoff

```javascript
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**:
```python
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**:
```json
{
  "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**:
1. Configure webhook URL in developer dashboard
2. Implement POST endpoint to receive notifications
3. Return 200 OK within 1 second
4. Process notification asynchronously

**Notification Payload**:
```json
{
  "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**:
```javascript
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

```python
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

1. **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

2. **Implement CDC or webhooks for syncing**
   - Avoid polling all entities repeatedly
   - CDC returns only changed entities
   - Webhooks provide real-time notifications without polling

3. **Sparse updates minimize payload**
   - Only send fields being changed
   - Reduces data transfer and processing time
   - Prevents accidental field overwrites

4. **Cache reference data locally**
   - Payment methods, tax codes, accounts rarely change
   - Query once and cache with TTL
   - Reduces redundant API calls

5. **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

1. **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

2. **Handle concurrent modifications gracefully**
   ```python
   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
   ```

3. **Validate required fields before API calls**
   - Check business rules locally first
   - Reduces validation errors from API
   - Example: Verify customer exists before creating invoice

4. **Use webhooks + CDC for reliable tracking**
   - Webhooks for real-time updates
   - Periodic CDC as backup for missed changes
   - Store last sync timestamp

### Token Management

1. **Access tokens expire after 3600 seconds**
   - Set up automatic refresh before expiration
   - Refresh at 50-minute mark to be safe

2. **Refresh tokens proactively**
   ```javascript
   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();
       }
     }
   }
   ```

3. **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

4. **Store tokens securely**
   - Encrypt in database
   - Never commit to version control
   - Use environment variables for development

5. **Handle 401 responses automatically**
   ```python
   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

1. **Implement exponential backoff for 429**
   ```python
   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
   ```

2. **Use batch operations to reduce call count**
   - 1 batch request vs 30 individual = 30x reduction
   - Batch counts as single API call for rate limits

3. **Monitor rate limit headers** (if provided)
   - Some endpoints return rate limit info in headers
   - Track usage to stay within limits

### Multi-currency Considerations

1. **CurrencyRef required when multicurrency enabled**
   ```json
   {
     "Invoice": {
       "CurrencyRef": {
         "value": "USD",
         "name": "United States Dollar"
       }
     }
   }
   ```

2. **Exchange rate handling**
   - API automatically applies exchange rates
   - ExchangeRate field shows conversion rate used
   - Home currency amounts calculated automatically

3. **Locale-specific required fields**
   - France: DocNumber required if custom transaction numbers enabled
   - UK: Different tax handling (VAT)
   - Check locale-specific documentation

### Testing & Development

1. **Use sandbox companies** (free with developer account)
   - Create at developer.intuit.com
   - Separate from production data
   - Full API feature parity

2. **Test OAuth flow end-to-end**
   - Authorization URL → code exchange → token refresh
   - Test token expiration handling
   - Verify refresh token rotation

3. **Validate webhook endpoint**
   - Test with sample payloads
   - Ensure < 1 second response time
   - Handle webhook signature verification

4. **Handle all fault types in production**
   - ValidationFault, SystemFault, AuthenticationFault, AuthorizationFault
   - Log error details (code, message, element)
   - Implement appropriate retry logic

5. **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**:

1. **Query or create customer**
```python
# 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']
```

2. **Query items for line items**
```python
# 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]
```

3. **Create invoice with line items**
```python
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']}")
```

4. **Send invoice email** (automatic if EmailStatus = "NeedToSend")
```python
# 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']}")
```

5. **Handle response and linked transactions**
```python
# 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**:

1. **Query invoice by DocNumber**
```python
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']
```

2. **Create payment entity**
```python
# 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'],
     

…(truncated)