Small and medium businesses lose an average of 14 hours per week on manual invoicing tasks ā reading client emails, extracting billing details, creating PDFs, and manually entering transactions into accounting software. An AI invoice automation agent can handle this entire workflow end-to-end: reading incoming emails, extracting invoice data with structured validation, generating professional PDF invoices, and logging transactions to your accounting ledger.
In this tutorial you will build a production-ready invoicing agent using LangChain, the Gmail API, reportlab for PDF generation, and a lightweight SQLite ledger for transaction logging. The same architecture extends to QuickBooks, Xero, or FreshBooks via their REST APIs.
Prerequisites#
Ensure you have the following before starting:
- Python 3.10 or later
- A Google Cloud project with Gmail API enabled
- Gmail OAuth credentials (
credentials.json) withgmail.readonlyandgmail.sendscopes - An OpenAI API key
- Basic Python knowledge including dataclasses
Install dependencies:
pip install langchain langchain-openai google-auth google-auth-oauthlib google-api-python-client reportlab pydantic python-dotenv
Architecture Overview#
The invoicing agent uses a linear tool chain triggered by incoming email events. The agent reads emails, determines if they are invoice requests, extracts structured data, generates a PDF, and logs the transaction.
Incoming Email (Gmail)
ā
ā¼
āāāāāāāāāāāāāāāāāāāāāāā
ā read_invoice_emails ā Gmail API ā Fetch unread emails labeled "invoice"
āāāāāāāāāāāā¬āāāāāāāāāāā
ā Raw email text
ā¼
āāāāāāāāāāāāāāāāāāāāāāā
ā extract_invoice ā LLM + Pydantic ā Structured invoice data
ā _data ā (client, amount, line items, due date)
āāāāāāāāāāāā¬āāāāāāāāāāā
ā InvoiceData object
ā¼
āāāāāāāāāāāāāāāāāāāāāāā
ā generate_pdf ā reportlab ā Invoice PDF saved to /invoices/
ā _invoice ā
āāāāāāāāāāāā¬āāāāāāāāāāā
ā PDF path
ā¼
āāāāāāāāāāāāāāāāāāāāāāā
ā log_to_ledger ā SQLite ā Transaction record with invoice ID
āāāāāāāāāāāā¬āāāāāāāāāāā
ā
ā¼
āāāāāāāāāāāāāāāāāāāāāāā
ā send_invoice_email ā Gmail API ā Email PDF to client
āāāāāāāāāāāāāāāāāāāāāāā
Step 1: Gmail API Setup#
The auth module follows the same pattern as the meeting scheduler tutorial, scoped specifically for Gmail:
# auth/gmail_auth.py
import os
from google.oauth2.credentials import Credentials
from google.auth.transport.requests import Request
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
SCOPES = [
"https://www.googleapis.com/auth/gmail.readonly",
"https://www.googleapis.com/auth/gmail.send",
"https://www.googleapis.com/auth/gmail.modify",
]
def get_gmail_service(token_path: str = "token_gmail.json",
creds_path: str = "credentials.json"):
creds = None
if os.path.exists(token_path):
creds = Credentials.from_authorized_user_file(token_path, SCOPES)
if not creds or not creds.valid:
if creds and creds.expired and creds.refresh_token:
creds.refresh(Request())
else:
flow = InstalledAppFlow.from_client_secrets_file(creds_path, SCOPES)
creds = flow.run_local_server(port=0)
with open(token_path, "w") as f:
f.write(creds.to_json())
return build("gmail", "v1", credentials=creds)
Create a Gmail label called invoice-requests in your Gmail account. The agent will only process emails with this label, preventing false positives from unrelated emails.
Step 2: Define the Invoice Data Model#
Use Pydantic to define a strict schema for extracted invoice data. This prevents the LLM from hallucinating fields and makes downstream processing reliable.
# models/invoice.py
from pydantic import BaseModel, Field, validator
from typing import Optional
from datetime import date, timedelta
class LineItem(BaseModel):
description: str = Field(..., description="Description of the product or service")
quantity: float = Field(..., gt=0, description="Quantity (must be positive)")
unit_price: float = Field(..., gt=0, description="Unit price in USD (must be positive)")
@property
def total(self) -> float:
return round(self.quantity * self.unit_price, 2)
class InvoiceData(BaseModel):
client_name: str = Field(..., description="Full name or company name of the client")
client_email: str = Field(..., description="Client's billing email address")
invoice_number: Optional[str] = Field(None, description="Invoice number if specified")
line_items: list[LineItem] = Field(..., min_items=1, description="Line items on the invoice")
due_date: Optional[str] = Field(None, description="Due date in YYYY-MM-DD format")
notes: Optional[str] = Field(None, description="Payment terms or special notes")
@validator("due_date", pre=True, always=True)
def set_default_due_date(cls, v):
if v is None:
return (date.today() + timedelta(days=30)).isoformat()
return v
@property
def subtotal(self) -> float:
return round(sum(item.total for item in self.line_items), 2)
@property
def tax_amount(self) -> float:
return round(self.subtotal * 0.0, 2) # Adjust tax rate as needed
@property
def total_due(self) -> float:
return round(self.subtotal + self.tax_amount, 2)
Step 3: Build the Email Reading Tool#
# tools/email_tools.py
import base64
from langchain.tools import tool
from auth.gmail_auth import get_gmail_service
@tool
def read_invoice_emails(max_results: int = 10) -> str:
"""
Fetch unread emails labeled 'invoice-requests' from Gmail.
Args:
max_results: Maximum number of emails to fetch (default 10).
Returns:
A formatted string with email IDs, subjects, senders, and body text.
"""
service = get_gmail_service()
results = service.users().messages().list(
userId="me",
labelIds=["UNREAD"],
q="label:invoice-requests",
maxResults=max_results,
).execute()
messages = results.get("messages", [])
if not messages:
return "No unread invoice request emails found."
email_summaries = []
for msg_ref in messages:
msg = service.users().messages().get(
userId="me",
id=msg_ref["id"],
format="full",
).execute()
headers = {h["name"]: h["value"] for h in msg["payload"]["headers"]}
subject = headers.get("Subject", "No subject")
sender = headers.get("From", "Unknown sender")
# Extract plain text body
body = ""
payload = msg["payload"]
if payload.get("body", {}).get("data"):
body = base64.urlsafe_b64decode(payload["body"]["data"]).decode("utf-8", errors="replace")
elif payload.get("parts"):
for part in payload["parts"]:
if part.get("mimeType") == "text/plain" and part.get("body", {}).get("data"):
body = base64.urlsafe_b64decode(part["body"]["data"]).decode("utf-8", errors="replace")
break
email_summaries.append(
f"EMAIL ID: {msg_ref['id']}\n"
f"FROM: {sender}\n"
f"SUBJECT: {subject}\n"
f"BODY:\n{body[:1000]}\n"
f"{'ā' * 60}"
)
return "\n\n".join(email_summaries)
Step 4: Build the Invoice Data Extraction Tool#
The extraction tool uses the LLM with structured output (via Pydantic) to parse free-form email text into the InvoiceData model:
# tools/extraction_tools.py
import json
from langchain.tools import tool
from langchain_openai import ChatOpenAI
from langchain.output_parsers import PydanticOutputParser
from langchain.prompts import ChatPromptTemplate
from models.invoice import InvoiceData
llm = ChatOpenAI(model="gpt-4o", temperature=0)
parser = PydanticOutputParser(pydantic_object=InvoiceData)
extraction_prompt = ChatPromptTemplate.from_messages([
("system", """You are an expert invoice data extractor.
Extract all invoice information from the email text provided.
{format_instructions}
If a field is not present in the email, use reasonable defaults.
For invoice_number, generate one in format INV-YYYYMMDD-001 if not specified."""),
("human", "Email text to extract invoice data from:\n\n{email_text}"),
])
@tool
def extract_invoice_data(email_text: str) -> str:
"""
Extract structured invoice data from an email using LLM parsing.
Args:
email_text: The raw email body text containing invoice request details.
Returns:
A JSON string of the extracted InvoiceData, or an error message.
"""
messages = extraction_prompt.format_messages(
format_instructions=parser.get_format_instructions(),
email_text=email_text,
)
response = llm.invoke(messages)
try:
invoice = parser.parse(response.content)
return invoice.json(indent=2)
except Exception as e:
return f"Extraction failed: {str(e)}\nRaw LLM output:\n{response.content}"
Step 5: Generate the PDF Invoice#
The PDF generator uses reportlab to produce a professional invoice document:
# tools/pdf_tools.py
import os
import json
from datetime import date
from langchain.tools import tool
from reportlab.lib.pagesizes import letter
from reportlab.lib import colors
from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
from reportlab.lib.units import inch
from reportlab.platypus import SimpleDocTemplate, Table, TableStyle, Paragraph, Spacer
from models.invoice import InvoiceData, LineItem
os.makedirs("invoices", exist_ok=True)
@tool
def generate_pdf_invoice(invoice_json: str) -> str:
"""
Generate a professional PDF invoice from structured invoice data.
Args:
invoice_json: JSON string of InvoiceData fields.
Returns:
The file path of the generated PDF, or an error message.
"""
try:
data = json.loads(invoice_json)
invoice = InvoiceData(**data)
except Exception as e:
return f"Failed to parse invoice data: {e}"
invoice_num = invoice.invoice_number or f"INV-{date.today().strftime('%Y%m%d')}-001"
filename = f"invoices/{invoice_num.replace('/', '-')}.pdf"
doc = SimpleDocTemplate(filename, pagesize=letter, rightMargin=72, leftMargin=72,
topMargin=72, bottomMargin=72)
styles = getSampleStyleSheet()
story = []
# Header
title_style = ParagraphStyle("Title", parent=styles["Heading1"], fontSize=24,
textColor=colors.HexColor("#1a1a2e"))
story.append(Paragraph("INVOICE", title_style))
story.append(Spacer(1, 0.2 * inch))
# Invoice meta
meta_data = [
["Invoice Number:", invoice_num],
["Date:", date.today().isoformat()],
["Due Date:", invoice.due_date],
["Bill To:", invoice.client_name],
["Email:", invoice.client_email],
]
meta_table = Table(meta_data, colWidths=[2 * inch, 4 * inch])
meta_table.setStyle(TableStyle([
("FONTNAME", (0, 0), (0, -1), "Helvetica-Bold"),
("FONTSIZE", (0, 0), (-1, -1), 11),
("BOTTOMPADDING", (0, 0), (-1, -1), 6),
]))
story.append(meta_table)
story.append(Spacer(1, 0.4 * inch))
# Line items table
header = ["Description", "Qty", "Unit Price", "Total"]
rows = [header]
for item in invoice.line_items:
rows.append([
item.description,
str(item.quantity),
f"${item.unit_price:,.2f}",
f"${item.total:,.2f}",
])
rows.append(["", "", "Subtotal:", f"${invoice.subtotal:,.2f}"])
rows.append(["", "", "Tax (0%):", f"${invoice.tax_amount:,.2f}"])
rows.append(["", "", "TOTAL DUE:", f"${invoice.total_due:,.2f}"])
items_table = Table(rows, colWidths=[3.5 * inch, 0.75 * inch, 1.25 * inch, 1 * inch])
items_table.setStyle(TableStyle([
("BACKGROUND", (0, 0), (-1, 0), colors.HexColor("#1a1a2e")),
("TEXTCOLOR", (0, 0), (-1, 0), colors.white),
("FONTNAME", (0, 0), (-1, 0), "Helvetica-Bold"),
("FONTSIZE", (0, 0), (-1, -1), 10),
("ROWBACKGROUNDS", (0, 1), (-1, -4), [colors.white, colors.HexColor("#f8f8f8")]),
("FONTNAME", (2, -1), (-1, -1), "Helvetica-Bold"),
("LINEBELOW", (0, 0), (-1, 0), 1, colors.HexColor("#1a1a2e")),
("LINEABOVE", (0, -3), (-1, -1), 0.5, colors.grey),
("ALIGN", (1, 0), (-1, -1), "RIGHT"),
("BOTTOMPADDING", (0, 0), (-1, -1), 8),
("TOPPADDING", (0, 0), (-1, -1), 8),
]))
story.append(items_table)
story.append(Spacer(1, 0.3 * inch))
if invoice.notes:
story.append(Paragraph(f"<b>Notes:</b> {invoice.notes}", styles["Normal"]))
doc.build(story)
return filename
Photo by Jakub Zerdzicki on Unsplash
Step 6: Log to Accounting Ledger#
The ledger tool writes transactions to a SQLite database, providing a simple audit trail that mirrors what you would send to QuickBooks or Xero:
# tools/ledger_tools.py
import sqlite3
import json
from datetime import date
from langchain.tools import tool
DB_PATH = "ledger.db"
def init_db():
conn = sqlite3.connect(DB_PATH)
conn.execute("""
CREATE TABLE IF NOT EXISTS transactions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
invoice_number TEXT NOT NULL,
client_name TEXT NOT NULL,
client_email TEXT NOT NULL,
total_due REAL NOT NULL,
due_date TEXT NOT NULL,
pdf_path TEXT,
status TEXT DEFAULT 'sent',
created_at TEXT NOT NULL
)
""")
conn.commit()
conn.close()
init_db()
@tool
def log_to_ledger(invoice_json: str, pdf_path: str) -> str:
"""
Log an invoice transaction to the accounting ledger database.
Args:
invoice_json: JSON string of InvoiceData fields.
pdf_path: File path to the generated PDF invoice.
Returns:
Confirmation with the database record ID.
"""
from models.invoice import InvoiceData
try:
data = json.loads(invoice_json)
invoice = InvoiceData(**data)
except Exception as e:
return f"Failed to parse invoice for ledger: {e}"
conn = sqlite3.connect(DB_PATH)
cursor = conn.execute(
"""INSERT INTO transactions
(invoice_number, client_name, client_email, total_due, due_date, pdf_path, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?)""",
(
invoice.invoice_number,
invoice.client_name,
invoice.client_email,
invoice.total_due,
invoice.due_date,
pdf_path,
date.today().isoformat(),
),
)
record_id = cursor.lastrowid
conn.commit()
conn.close()
return (
f"Transaction logged successfully.\n"
f"Record ID: {record_id}\n"
f"Invoice: {invoice.invoice_number}\n"
f"Client: {invoice.client_name}\n"
f"Amount: ${invoice.total_due:,.2f}\n"
f"Due: {invoice.due_date}"
)
Step 7: Assemble the Invoicing Agent#
# agent.py
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from tools.email_tools import read_invoice_emails
from tools.extraction_tools import extract_invoice_data
from tools.pdf_tools import generate_pdf_invoice
from tools.ledger_tools import log_to_ledger
load_dotenv()
llm = ChatOpenAI(model="gpt-4o", temperature=0)
tools = [
read_invoice_emails,
extract_invoice_data,
generate_pdf_invoice,
log_to_ledger,
]
system_prompt = """You are an accounts receivable automation agent. Your job is to:
1. Read unread invoice request emails using read_invoice_emails.
2. For each email, extract structured invoice data using extract_invoice_data.
3. Generate a PDF invoice using generate_pdf_invoice.
4. Log the transaction to the ledger using log_to_ledger.
5. Provide a summary of all invoices processed.
Process ALL unread invoice emails. After completing each invoice, move to the next.
Report any failures clearly so they can be handled manually.
"""
prompt = ChatPromptTemplate.from_messages([
("system", system_prompt),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad"),
])
agent = create_openai_tools_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
max_iterations=20, # Allow multiple iterations for batch processing
handle_parsing_errors=True,
)
def run_invoice_automation() -> str:
result = agent_executor.invoke({
"input": "Process all unread invoice request emails. "
"Extract data, generate PDFs, log to ledger, and provide a summary."
})
return result["output"]
if __name__ == "__main__":
summary = run_invoice_automation()
print("\n--- INVOICE AUTOMATION SUMMARY ---")
print(summary)
Testing the Agent#
# test_invoicing.py
import pytest
import json
from models.invoice import InvoiceData, LineItem
def test_invoice_model_calculates_totals():
"""InvoiceData should correctly calculate subtotal and total."""
invoice = InvoiceData(
client_name="Acme Corp",
client_email="billing@acme.com",
invoice_number="INV-20260225-001",
line_items=[
LineItem(description="API Development", quantity=10, unit_price=150.0),
LineItem(description="Code Review", quantity=2, unit_price=200.0),
],
due_date="2026-03-25",
)
assert invoice.subtotal == 1900.0
assert invoice.total_due == 1900.0 # 0% tax in this config
def test_line_item_total():
"""LineItem total should equal quantity * unit_price."""
item = LineItem(description="Consulting", quantity=5, unit_price=120.0)
assert item.total == 600.0
def test_default_due_date_is_30_days():
"""Due date should default to 30 days from today if not specified."""
from datetime import date, timedelta
invoice = InvoiceData(
client_name="Test Client",
client_email="test@test.com",
line_items=[LineItem(description="Service", quantity=1, unit_price=100.0)],
)
expected = (date.today() + timedelta(days=30)).isoformat()
assert invoice.due_date == expected
Run: pytest test_invoicing.py -v
Production Considerations#
Scheduling: Run the agent on a cron schedule using schedule or a cloud scheduler:
import schedule
import time
schedule.every().hour.do(run_invoice_automation)
while True:
schedule.run_pending()
time.sleep(60)
Email threading: After generating the invoice, send it back in the same Gmail thread by including the threadId from the original email in the send request. This keeps client communication organized.
Connecting to QuickBooks Online: Replace the SQLite ledger with the QuickBooks Python SDK (intuit-oauth, python-quickbooks) to post invoices directly to your QBO account:
from quickbooks import QuickBooks
from quickbooks.objects.invoice import Invoice as QBOInvoice
client = QuickBooks(
client_id=os.getenv("QBO_CLIENT_ID"),
client_secret=os.getenv("QBO_CLIENT_SECRET"),
refresh_token=os.getenv("QBO_REFRESH_TOKEN"),
environment="production",
company_id=os.getenv("QBO_COMPANY_ID"),
)
# Map InvoiceData ā QBO Invoice object and save
Fraud detection: Add a validation step that flags invoices above a threshold (e.g., $10,000) for human review before PDF generation and sending. Use a simple rule in the agent's system prompt: "If total_due exceeds $10,000, stop and request human approval."
Idempotency: Track processed email IDs in your database to avoid processing the same invoice twice if the agent runs before an email is marked as read.
Further Reading#
- Getting Started with AI Agents ā Foundational LangChain agent concepts
- How to Build a Meeting Scheduler AI Agent ā Gmail API setup patterns you can reuse here
- How to Build a Content Writing AI Agent ā Multi-agent patterns with CrewAI
- AI Agents for Finance Use Cases ā Discover more finance automation opportunities
- LangChain Tool Calling Guide ā Deep dive into the tool architecture used in this tutorial
- AI Agent Integration Guides ā Connect your invoicing agent to accounting platforms
Frequently Asked Questions#
Is it safe to give an AI agent access to my Gmail inbox?
Yes, when done correctly. The OAuth scopes in this tutorial request only gmail.readonly and gmail.send ā the agent cannot delete emails or modify inbox state beyond marking processed emails as read. Use a dedicated Gmail label (invoice-requests) to further limit the blast radius. Never grant broad Gmail access scopes in production.
How do I handle invoices in multiple currencies?
Add a currency field to the InvoiceData model and update the PDF generator to display the correct currency symbol. For exchange rate conversion, integrate with a free API like Open Exchange Rates and add a currency conversion tool to the agent.
Can this integrate with Stripe for payment collection?
Yes. After generating the PDF, call the Stripe API to create a Payment Intent or a Stripe-hosted Invoice, then include the payment link in the confirmation email. Add a create_stripe_invoice tool that wraps the Stripe Python SDK.
What happens if the LLM extracts incorrect invoice data? The Pydantic model catches structural errors (wrong types, missing required fields) and returns an error rather than generating an incorrect invoice. For semantic errors (wrong amounts due to ambiguous email language), add a confirmation step: have the agent send a brief email to the requester asking them to confirm the extracted total before generating the PDF.