quickbooks-online

What this skill does

# QuickBooks Online

## Overview

Connect your application to QuickBooks Online (QBO) via the Intuit Developer API. This skill covers OAuth2 setup, CRUD operations for core accounting entities (invoices, bills, payments, customers, vendors, accounts), and report generation (P&L, balance sheet, cash flow). Works with both sandbox and production QBO companies.

## Instructions

### Step 1: Set up OAuth2

Register your app at [developer.intuit.com](https://developer.intuit.com). QBO uses OAuth2 with refresh tokens (expire after 100 days).

```python
# qbo_auth.py — OAuth2 token management for QuickBooks Online
import requests
import base64
import json
import os
from datetime import datetime, timedelta

QBO_CLIENT_ID = os.environ["QBO_CLIENT_ID"]
QBO_CLIENT_SECRET = os.environ["QBO_CLIENT_SECRET"]
QBO_REDIRECT_URI = os.environ["QBO_REDIRECT_URI"]
QBO_ENVIRONMENT = os.environ.get("QBO_ENVIRONMENT", "sandbox")  # "sandbox" or "production"

TOKEN_FILE = ".qbo_tokens.json"

BASE_URL = {
    "sandbox": "https://sandbox-quickbooks.api.intuit.com",
    "production": "https://quickbooks.api.intuit.com",
}[QBO_ENVIRONMENT]

def get_auth_url() -> str:
    """Generate the OAuth2 authorization URL."""
    import secrets
    state = secrets.token_urlsafe(16)
    params = {
        "client_id": QBO_CLIENT_ID,
        "scope": "com.intuit.quickbooks.accounting",
        "redirect_uri": QBO_REDIRECT_URI,
        "response_type": "code",
        "state": state,
    }
    query = "&".join(f"{k}={v}" for k, v in params.items())
    return f"https://appcenter.intuit.com/connect/oauth2?{query}"

def exchange_code_for_tokens(code: str, realm_id: str) -> dict:
    """Exchange authorization code for tokens. realm_id is the QBO company ID."""
    credentials = base64.b64encode(
        f"{QBO_CLIENT_ID}:{QBO_CLIENT_SECRET}".encode()
    ).decode()

    response = requests.post(
        "https://oauth.platform.intuit.com/oauth2/v1/tokens/bearer",
        headers={
            "Authorization": f"Basic {credentials}",
            "Content-Type": "application/x-www-form-urlencoded",
        },
        data={
            "grant_type": "authorization_code",
            "code": code,
            "redirect_uri": QBO_REDIRECT_URI,
        },
    )
    tokens = response.json()
    tokens["realm_id"] = realm_id
    tokens["expires_at"] = (
        datetime.utcnow() + timedelta(seconds=tokens["expires_in"])
    ).isoformat()
    tokens["refresh_token_expires_at"] = (
        datetime.utcnow() + timedelta(days=100)
    ).isoformat()
    save_tokens(tokens)
    return tokens

def refresh_access_token() -> dict:
    """Refresh the access token (expires after 1 hour)."""
    tokens = load_tokens()
    credentials = base64.b64encode(
        f"{QBO_CLIENT_ID}:{QBO_CLIENT_SECRET}".encode()
    ).decode()

    response = requests.post(
        "https://oauth.platform.intuit.com/oauth2/v1/tokens/bearer",
        headers={
            "Authorization": f"Basic {credentials}",
            "Content-Type": "application/x-www-form-urlencoded",
        },
        data={
            "grant_type": "refresh_token",
            "refresh_token": tokens["refresh_token"],
        },
    )
    new_tokens = response.json()
    new_tokens["realm_id"] = tokens["realm_id"]
    new_tokens["expires_at"] = (
        datetime.utcnow() + timedelta(seconds=new_tokens["expires_in"])
    ).isoformat()
    new_tokens["refresh_token_expires_at"] = tokens["refresh_token_expires_at"]
    save_tokens(new_tokens)
    return new_tokens

def get_valid_token() -> tuple[str, str]:
    """Return (access_token, realm_id), refreshing if needed."""
    tokens = load_tokens()
    expires_at = datetime.fromisoformat(tokens["expires_at"])
    if datetime.utcnow() >= expires_at - timedelta(minutes=5):
        tokens = refresh_access_token()
    return tokens["access_token"], tokens["realm_id"]

def save_tokens(tokens: dict):
    with open(TOKEN_FILE, "w") as f:
        json.dump(tokens, f)

def load_tokens() -> dict:
    with open(TOKEN_FILE) as f:
        return json.load(f)
```

### Step 2: Create an API client

```python
# qbo_client.py — QBO API client using the v3 REST API
import requests
from qbo_auth import get_valid_token, BASE_URL

class QBOClient:
    def __init__(self):
        self.token, self.realm_id = get_valid_token()
        self.base = f"{BASE_URL}/v3/company/{self.realm_id}"

    def _headers(self):
        return {
            "Authorization": f"Bearer {self.token}",
            "Accept": "application/json",
            "Content-Type": "application/json",
        }

    def query(self, sql: str) -> list:
        """Run a QBO query (SQL-like syntax)."""
        response = requests.get(
            f"{self.base}/query",
            headers=self._headers(),
            params={"query": sql, "minorversion": "65"},
        )
        response.raise_for_status()
        data = response.json()
        # The response key varies by entity: QueryResponse.Invoice, .Customer, etc.
        query_response = data.get("QueryResponse", {})
        for key, value in query_response.items():
            if isinstance(value, list):
                return value
        return []

    def create(self, entity: str, payload: dict) -> dict:
        """Create a QBO entity."""
        response = requests.post(
            f"{self.base}/{entity}?minorversion=65",
            headers=self._headers(),
            json=payload,
        )
        response.raise_for_status()
        return response.json()

    def update(self, entity: str, payload: dict) -> dict:
        """Update a QBO entity (requires Id and SyncToken in payload)."""
        response = requests.post(
            f"{self.base}/{entity}?operation=update&minorversion=65",
            headers=self._headers(),
            json=payload,
        )
        response.raise_for_status()
        return response.json()

    def get_report(self, report_name: str, params: dict = None) -> dict:
        """Fetch a financial report."""
        response = requests.get(
            f"{self.base}/reports/{report_name}",
            headers=self._headers(),
            params={**(params or {}), "minorversion": "65"},
        )
        response.raise_for_status()
        return response.json()
```

### Step 3: Manage customers and vendors

```python
# entities.py — Customers and vendors
from qbo_client import QBOClient

client = QBOClient()

def upsert_customer(display_name: str, email: str = None, phone: str = None,
                    company_name: str = None) -> dict:
    """Find or create a customer in QBO."""
    # Try to find existing customer
    results = client.query(
        f"SELECT * FROM Customer WHERE DisplayName = '{display_name}'"
    )
    if results:
        return results[0]

    payload = {"DisplayName": display_name}
    if email:
        payload["PrimaryEmailAddr"] = {"Address": email}
    if phone:
        payload["PrimaryPhone"] = {"FreeFormNumber": phone}
    if company_name:
        payload["CompanyName"] = company_name

    result = client.create("customer", payload)
    return result["Customer"]

def upsert_vendor(display_name: str, email: str = None, phone: str = None) -> dict:
    """Find or create a vendor in QBO."""
    results = client.query(
        f"SELECT * FROM Vendor WHERE DisplayName = '{display_name}'"
    )
    if results:
        return results[0]

    payload = {"DisplayName": display_name}
    if email:
        payload["PrimaryEmailAddr"] = {"Address": email}

    result = client.create("vendor", payload)
    return result["Vendor"]
```

### Step 4: Create and manage invoices

```python
# invoices.py — Invoice operations
from qbo_client import QBOClient
from entities import upsert_customer
from datetime import datetime, timedelta

client = QBOClient()

def create_invoice(customer_name: str, line_items: list,
                   due_days: int = 30, currency: str = "USD") -> dict:
    """
    Create an invoice in QBO.

    line_items format:
      [{"description": ".