How to Integrate AI Agents with Xero

pip install requests langchain langchain-openai python-dotenv

export XERO_CLIENT_ID="your-client-id"
export XERO_CLIENT_SECRET="your-client-secret"
export XERO_REFRESH_TOKEN="your-refresh-token"
export XERO_TENANT_ID="your-tenant-id"   # Organisation ID from /connections

import os, requests

CLIENT_ID = os.getenv("XERO_CLIENT_ID")
CLIENT_SECRET = os.getenv("XERO_CLIENT_SECRET")
REFRESH_TOKEN = os.getenv("XERO_REFRESH_TOKEN")
TENANT_ID = os.getenv("XERO_TENANT_ID")

def get_xero_token() -> str:
    resp = requests.post(
        "https://identity.xero.com/connect/token",
        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_xero_token()
resp = requests.get(
    "https://api.xero.com/api.xro/2.0/Organisation",
    headers={"Authorization": f"Bearer {token}", "Xero-tenant-id": TENANT_ID,
             "Accept": "application/json"}
)
org = resp.json()["Organisations"][0]
print(f"Connected to: {org['Name']} ({org['CountryCode']})")

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("XERO_CLIENT_ID")
CLIENT_SECRET = os.getenv("XERO_CLIENT_SECRET")
REFRESH_TOKEN = os.getenv("XERO_REFRESH_TOKEN")
TENANT_ID = os.getenv("XERO_TENANT_ID")
XERO_BASE = "https://api.xero.com/api.xro/2.0"


def get_xero_token() -> str:
    """Exchange refresh token for fresh access token."""
    resp = requests.post(
        "https://identity.xero.com/connect/token",
        auth=(CLIENT_ID, CLIENT_SECRET),
        data={"grant_type": "refresh_token", "refresh_token": REFRESH_TOKEN}
    )
    resp.raise_for_status()
    return resp.json()["access_token"]


def xero_get(endpoint: str, params: dict = None) -> dict:
    """Execute an authenticated Xero API GET request."""
    token = get_xero_token()
    headers = {
        "Authorization": f"Bearer {token}",
        "Xero-tenant-id": TENANT_ID,
        "Accept": "application/json"
    }
    resp = requests.get(f"{XERO_BASE}/{endpoint}", headers=headers, params=params or {})
    resp.raise_for_status()
    return resp.json()


def xero_report(report_name: str, params: dict = None) -> dict:
    """Fetch a Xero financial report."""
    token = get_xero_token()
    headers = {
        "Authorization": f"Bearer {token}",
        "Xero-tenant-id": TENANT_ID,
        "Accept": "application/json"
    }
    resp = requests.get(
        f"{XERO_BASE}/Reports/{report_name}",
        headers=headers, params=params or {}
    )
    resp.raise_for_status()
    return resp.json()


@tool
def get_profit_loss(from_date: str, to_date: str) -> str:
    """
    Get Profit & Loss report for a date range.
    date format: 'YYYY-MM-DD'
    """
    data = xero_report("ProfitAndLoss", {
        "fromDate": from_date,
        "toDate": to_date,
        "standardLayout": "true"
    })
    reports = data.get("Reports", [])
    if not reports:
        return "No P&L data available"

    report = reports[0]
    lines = [f"Profit & Loss ({from_date} to {to_date}):"]

    for section in report.get("Rows", []):
        row_type = section.get("RowType", "")
        if row_type == "Section":
            title = section.get("Title", "")
            if title:
                lines.append(f"\n{title}:")
            for row in section.get("Rows", []):
                cells = row.get("Cells", [])
                if len(cells) >= 2:
                    label = cells[0].get("Value", "")
                    amount = cells[1].get("Value", "0")
                    if label and amount and amount != "":
                        try:
                            lines.append(f"  {label}: ${float(amount):,.2f}")
                        except ValueError:
                            pass
        elif row_type == "SummaryRow":
            cells = section.get("Cells", [])
            if len(cells) >= 2:
                label = cells[0].get("Value", "")
                amount = cells[1].get("Value", "0")
                try:
                    lines.append(f"\n{label}: ${float(amount):,.2f}")
                except ValueError:
                    pass

    return "\n".join(lines)


@tool
def get_outstanding_invoices(min_days_overdue: int = 0) -> str:
    """
    Get unpaid invoices from Xero ranked by days overdue.
    min_days_overdue: only show invoices overdue by at least this many days.
    """
    data = xero_get("Invoices", {
        "Status": "AUTHORISED",
        "where": "AmountDue>0",
        "order": "DueDate ASC"
    })
    invoices = data.get("Invoices", [])

    if not invoices:
        return "No outstanding invoices found"

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

    for inv in invoices[:50]:
        contact_name = inv.get("Contact", {}).get("Name", "Unknown")
        amount_due = float(inv.get("AmountDue", 0))
        due_date_str = inv.get("DueDateString", "")
        inv_number = inv.get("InvoiceNumber", "N/A")
        days_overdue = 0

        if due_date_str:
            try:
                due_date = datetime.strptime(due_date_str[:10], "%Y-%m-%d").date()
                days_overdue = (today - due_date).days
            except ValueError:
                pass

        if days_overdue >= min_days_overdue:
            status = f"{days_overdue}d overdue" if days_overdue > 0 else f"due {due_date_str[:10]}"
            lines.append(f"  {inv_number} | {contact_name} | ${amount_due:,.2f} | {status}")
            total += amount_due

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

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


@tool
def get_bank_account_balances() -> str:
    """Get current balances for all Xero bank accounts."""
    data = xero_get("Accounts", {"Type": "BANK", "where": "Status==\"ACTIVE\""})
    accounts = data.get("Accounts", [])

    if not accounts:
        return "No bank accounts found"

    lines = ["Bank account balances:"]
    total = 0.0
    for account in accounts:
        name = account.get("Name", "Unknown")
        balance = float(account.get("CurrencyCode", "0") and
                       account.get("ReportingCodeUpdatedDateUTC", "0") and
                       "0")  # Balance requires separate API call
        code = account.get("Code", "")
        currency = account.get("CurrencyCode", "")
        lines.append(f"  {name} ({code}): {currency} — use Bank Summary report for balances")

    # Get actual balances from BankSummary report
    try:
        summary = xero_report("BankSummary")
        reports = summary.get("Reports", [])
        if reports:
            lines = ["Bank account balances:"]
            total = 0.0
            for row in reports[0].get("Rows", []):
                cells = row.get("Cells", [])
                if len(cells) >= 5:
                    account_name = cells[0].get("Value", "")
                    closing_balance = cells[4].get("Value", "0")
                    if account_name and closing_balance:
                        try:
                            bal = float(closing_balance)
                            lines.append(f"  {account_name}: ${bal:,.2f}")
                            total += bal
                        except ValueError:
                            pass
            lines.append(f"\nTotal cash: ${total:,.2f}")
    except Exception:
        pass

    return "\n".join(lines)


@tool
def get_aged_receivables() -> str:
    """Get aged receivables report showing outstanding amounts by aging bucket."""
    data = xero_report("AgedReceivablesByContact")
    reports = data.get("Reports", [])
    if not reports:
        return "No aged receivables data available"

    lines = ["Aged Receivables Summary:"]
    report = reports[0]

    for row in report.get("Rows", []):
        if row.get("RowType") == "Row":
            cells = row.get("Cells", [])
            if len(cells) >= 6:
                contact = cells[0].get("Value", "")
                current = cells[1].get("Value", "0")
                days_30 = cells[2].get("Value", "0")
                days_60 = cells[3].get("Value", "0")
                days_90 = cells[4].get("Value", "0")
                total = cells[5].get("Value", "0")
                try:
                    if float(total) > 0:
                        lines.append(f"  {contact}: Current ${float(current):,.0f} | "
                                     f"30d ${float(days_30):,.0f} | "
                                     f"60d ${float(days_60):,.0f} | "
                                     f"90d+ ${float(days_90):,.0f} | "
                                     f"Total ${float(total):,.0f}")
                except ValueError:
                    pass

    return "\n".join(lines) if len(lines) > 1 else "No outstanding receivables"


@tool
def get_expense_claims() -> str:
    """Get pending employee expense claims awaiting approval."""
    data = xero_get("ExpenseClaims", {"Status": "SUBMITTED"})
    claims = data.get("ExpenseClaims", [])

    if not claims:
        return "No pending expense claims"

    lines = [f"Pending expense claims ({len(claims)}):"]
    total = 0.0
    for claim in claims:
        user = claim.get("User", {})
        name = f"{user.get('FirstName', '')} {user.get('LastName', '')}".strip()
        amount = float(claim.get("AmountDue", 0))
        status = claim.get("Status", "")
        total += amount
        lines.append(f"  {name} | ${amount:,.2f} | {status}")

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

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_bank_account_balances,
         get_aged_receivables, get_expense_claims]

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

When answering financial questions:
1. Specify date ranges clearly for all reports
2. Rank overdue invoices by urgency (90+ days is critical, 60+ is high priority)
3. Present currency values with $ and comma formatting: $12,345.67
4. Highlight patterns that suggest cash flow risk or collection problems
5. Recommend one specific action based on the data"""),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

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