How to Integrate AI Agents with QuickBooks

pip install requests langchain langchain-openai python-dotenv

export QUICKBOOKS_CLIENT_ID="your-client-id"
export QUICKBOOKS_CLIENT_SECRET="your-client-secret"
export QUICKBOOKS_REALM_ID="your-company-id"       # Appears in QBO URL
export QUICKBOOKS_REFRESH_TOKEN="your-refresh-token"

import os, requests

CLIENT_ID = os.getenv("QUICKBOOKS_CLIENT_ID")
CLIENT_SECRET = os.getenv("QUICKBOOKS_CLIENT_SECRET")
REFRESH_TOKEN = os.getenv("QUICKBOOKS_REFRESH_TOKEN")
REALM_ID = os.getenv("QUICKBOOKS_REALM_ID")

def get_access_token() -> str:
    resp = requests.post(
        "https://oauth.platform.intuit.com/oauth2/v1/tokens/bearer",
        auth=(CLIENT_ID, CLIENT_SECRET),
        data={"grant_type": "refresh_token", "refresh_token": REFRESH_TOKEN}
    )
    resp.raise_for_status()
    return resp.json()["access_token"]

token = get_access_token()
resp = requests.get(
    f"https://quickbooks.api.intuit.com/v3/company/{REALM_ID}/companyinfo/{REALM_ID}",
    headers={"Authorization": f"Bearer {token}", "Accept": "application/json"},
    params={"minorversion": "70"}
)
print(f"Connected to: {resp.json()['CompanyInfo']['CompanyName']}")

import os
import requests
from datetime import datetime, timedelta
from langchain.tools import tool
from dotenv import load_dotenv

load_dotenv()

CLIENT_ID = os.getenv("QUICKBOOKS_CLIENT_ID")
CLIENT_SECRET = os.getenv("QUICKBOOKS_CLIENT_SECRET")
REALM_ID = os.getenv("QUICKBOOKS_REALM_ID")
REFRESH_TOKEN = os.getenv("QUICKBOOKS_REFRESH_TOKEN")
QB_BASE = f"https://quickbooks.api.intuit.com/v3/company/{REALM_ID}"


def get_access_token() -> str:
    """Exchange refresh token for a fresh access token."""
    resp = requests.post(
        "https://oauth.platform.intuit.com/oauth2/v1/tokens/bearer",
        auth=(CLIENT_ID, CLIENT_SECRET),
        data={"grant_type": "refresh_token", "refresh_token": REFRESH_TOKEN}
    )
    resp.raise_for_status()
    return resp.json()["access_token"]


def qb_request(method: str, path: str, params: dict = None, json_data: dict = None) -> dict:
    """Execute a QuickBooks API request with fresh token authentication."""
    token = get_access_token()
    headers = {"Authorization": f"Bearer {token}", "Accept": "application/json",
               "Content-Type": "application/json"}
    params = {**(params or {}), "minorversion": "70"}
    resp = requests.request(method, f"{QB_BASE}/{path}", headers=headers,
                            params=params, json=json_data)
    resp.raise_for_status()
    return resp.json()


def qb_report(report_name: str, params: dict = None) -> dict:
    """Fetch a QuickBooks financial report by name."""
    token = get_access_token()
    headers = {"Authorization": f"Bearer {token}", "Accept": "application/json"}
    params = {**(params or {}), "minorversion": "70"}
    resp = requests.get(
        f"https://quickbooks.api.intuit.com/v3/company/{REALM_ID}/reports/{report_name}",
        headers=headers, params=params
    )
    resp.raise_for_status()
    return resp.json()


def parse_report_rows(report: dict) -> list:
    """Parse QuickBooks report row structure into flat key-value pairs."""
    results = []
    for row in report.get("Rows", {}).get("Row", []):
        header = row.get("Header", {}).get("ColData", [])
        title = header[0].get("value", "") if header else ""
        summary = row.get("Summary", {}).get("ColData", [])
        amount = summary[-1].get("value", "0") if summary else "0"
        try:
            amount_val = float(amount)
            if title and amount_val != 0:
                results.append((title, amount_val))
        except ValueError:
            pass
    return results


@tool
def get_profit_loss(start_date: str, end_date: str) -> str:
    """
    Get Profit & Loss report for a date range.
    date format: 'YYYY-MM-DD'
    """
    report = qb_report("ProfitAndLoss", {
        "start_date": start_date,
        "end_date": end_date,
        "accounting_method": "Accrual"
    })
    rows = parse_report_rows(report)
    if not rows:
        return "No P&L data available for this date range"

    lines = [f"Profit & Loss ({start_date} to {end_date}):"]
    for title, amount in rows:
        prefix = "  " if "Net" not in title else ""
        lines.append(f"{prefix}{title}: ${amount:,.2f}")
    return "\n".join(lines)


@tool
def get_outstanding_invoices(min_days_overdue: int = 0) -> str:
    """
    Get unpaid invoices from QuickBooks ranked by days overdue.
    min_days_overdue: filter to only show invoices past due by at least this many days.
    """
    result = qb_request("GET", "query", {
        "query": "SELECT * FROM Invoice WHERE Balance > '0' MAXRESULTS 50 ORDERBY DueDate ASC"
    })
    entities = result.get("QueryResponse", {}).get("Invoice", [])

    if not entities:
        return "No outstanding invoices found"

    today = datetime.today().date()
    lines = []
    total = 0

    for inv in entities:
        customer = inv.get("CustomerRef", {}).get("name", "Unknown")
        balance = float(inv.get("Balance", 0))
        due_str = inv.get("DueDate", "")
        doc_num = inv.get("DocNumber", "N/A")
        days_overdue = 0

        if due_str:
            due_date = datetime.strptime(due_str, "%Y-%m-%d").date()
            days_overdue = (today - due_date).days

        if days_overdue >= min_days_overdue:
            status = f"{days_overdue}d overdue" if days_overdue > 0 else f"due {due_str}"
            lines.append(f"  #{doc_num} | {customer} | ${balance:,.2f} | {status}")
            total += balance

    if not lines:
        return f"No invoices overdue by {min_days_overdue}+ days"

    header = f"Outstanding invoices ({len(lines)} found):"
    footer = f"\nTotal AR outstanding: ${total:,.2f}"
    return header + "\n" + "\n".join(lines) + footer


@tool
def get_accounts_payable() -> str:
    """Get outstanding bills (accounts payable) grouped by vendor with due dates."""
    result = qb_request("GET", "query", {
        "query": "SELECT * FROM Bill WHERE Balance > '0' MAXRESULTS 50 ORDERBY DueDate ASC"
    })
    entities = result.get("QueryResponse", {}).get("Bill", [])

    if not entities:
        return "No outstanding bills found"

    today = datetime.today().date()
    lines = [f"Accounts payable ({len(entities)} bills):"]
    total = 0

    for bill in entities:
        vendor = bill.get("VendorRef", {}).get("name", "Unknown")
        balance = float(bill.get("Balance", 0))
        due_str = bill.get("DueDate", "")
        days_until = ""
        if due_str:
            due_date = datetime.strptime(due_str, "%Y-%m-%d").date()
            days_diff = (due_date - today).days
            days_until = f" ({abs(days_diff)}d {'overdue' if days_diff < 0 else 'remaining'})"
        total += balance
        lines.append(f"  {vendor} | ${balance:,.2f} | Due: {due_str}{days_until}")

    lines.append(f"\nTotal AP: ${total:,.2f}")
    return "\n".join(lines)


@tool
def get_cash_flow_snapshot() -> str:
    """Get a cash flow summary for the past 30 days."""
    end = datetime.today().date()
    start = end - timedelta(days=30)

    report = qb_report("CashFlow", {
        "start_date": start.isoformat(),
        "end_date": end.isoformat()
    })
    rows = parse_report_rows(report)
    if not rows:
        return "No cash flow data available"

    lines = [f"Cash Flow ({start} to {end}):"]
    for title, amount in rows:
        lines.append(f"  {title}: ${amount:,.2f}")
    return "\n".join(lines)


@tool
def get_expense_summary(start_date: str, end_date: str) -> str:
    """Get expenses summarized by account category for a date range."""
    report = qb_report("ProfitAndLoss", {
        "start_date": start_date,
        "end_date": end_date,
        "accounting_method": "Cash"
    })
    lines = [f"Expense summary ({start_date} to {end_date}):"]
    for row in report.get("Rows", {}).get("Row", []):
        header = row.get("Header", {}).get("ColData", [])
        title = header[0].get("value", "") if header else ""
        if "Expense" in title or "Cost" in title or "Overhead" in title:
            for sub_row in row.get("Rows", {}).get("Row", []):
                sub_header = sub_row.get("Header", {}).get("ColData", [])
                sub_title = sub_header[0].get("value", "") if sub_header else ""
                sub_summary = sub_row.get("Summary", {}).get("ColData", [])
                sub_amount = sub_summary[-1].get("value", "0") if sub_summary else "0"
                try:
                    amt = float(sub_amount)
                    if sub_title and amt != 0:
                        lines.append(f"  {sub_title}: ${amt:,.2f}")
                except ValueError:
                    pass
    return "\n".join(lines) if len(lines) > 1 else "No expense data found"

from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o", temperature=0)
tools = [get_profit_loss, get_outstanding_invoices, get_accounts_payable,
         get_cash_flow_snapshot, get_expense_summary]

prompt = ChatPromptTemplate.from_messages([
    ("system", """You are a financial operations assistant with access to QuickBooks Online.

When answering financial questions:
1. Always specify the date range for reports
2. Flag AR items by overdue severity: current, 30-60 days, 60-90 days, 90+ days (most urgent)
3. Highlight unusual expense categories or amounts compared to typical business patterns
4. Translate financial metrics into operational insights, not just numbers
5. Never speculate about future performance — only report on available data
6. Always format monetary values with dollar signs and commas: $12,345.67"""),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True, max_iterations=6)

# Financial health check
result = executor.invoke({
    "input": "Give me a financial snapshot: this month's P&L, how much AR is overdue and by how long, and current AP balance."
})
print(result["output"])