"""
Expense management tools.
Provides CRUD operations for expenses including attachments.
"""
from __future__ import annotations
import base64
from pathlib import Path
from typing import TYPE_CHECKING
from ..constants import (
EXPENSE_STATE_LABELS,
VALID_EXPENSE_STATES,
ExpenseState,
OdooModel,
PaymentMode,
)
from ..decorators import handle_odoo_errors
from ..exceptions import OdooValidationError
from ..formatters import MarkdownBuilder, format_money
from ..validators import validate_date, validate_positive_amount, today
from .base import extract_name, format_state, get_odoo_client, normalize_pagination
if TYPE_CHECKING:
from mcp.server.fastmcp import FastMCP
def register_tools(mcp: "FastMCP") -> None:
"""Register expense tools with the MCP server."""
@mcp.tool()
@handle_odoo_errors
def list_expense_categories(
limit: int = 100,
offset: int = 0,
) -> str:
"""
List available expense categories (service type products).
Args:
limit: Maximum number of categories to return (default: 100)
offset: Offset for pagination (default: 0)
Returns:
List of categories with their ID and name
"""
limit, offset = normalize_pagination(limit, offset)
client = get_odoo_client()
categories = client.search_read(
OdooModel.PRODUCT,
[("can_be_expensed", "=", True)],
["name", "standard_price", "list_price"],
limit=limit,
offset=offset,
order="name asc",
)
if not categories:
return "No expense categories found."
builder = MarkdownBuilder("Expense Categories")
rows = []
for cat in categories:
rows.append([
cat["id"],
cat["name"],
format_money(cat.get("standard_price", 0)),
])
builder.add_table(
["ID", "Name", "Cost"],
rows,
alignments=["right", "left", "right"],
)
builder.add_pagination(len(categories), limit, offset)
return builder.build()
@mcp.tool()
@handle_odoo_errors
def list_expenses(
date_from: str | None = None,
date_to: str | None = None,
state: str | None = None,
limit: int = 50,
offset: int = 0,
) -> str:
"""
List existing expenses.
Args:
date_from: Start date (format YYYY-MM-DD, optional)
date_to: End date (format YYYY-MM-DD, optional)
state: Filter by state: draft, reported, approved, done, refused (optional)
limit: Maximum number of entries (default: 50)
offset: Offset for pagination (default: 0)
Returns:
List of expenses
"""
limit, offset = normalize_pagination(limit, offset)
client = get_odoo_client()
domain: list = []
if date_from:
domain.append(("date", ">=", validate_date(date_from, "date_from")))
if date_to:
domain.append(("date", "<=", validate_date(date_to, "date_to")))
if state:
if state not in VALID_EXPENSE_STATES:
raise OdooValidationError(
f"Invalid state: {state}. Valid: {', '.join(VALID_EXPENSE_STATES)}",
field="state",
)
domain.append(("state", "=", state))
expenses = client.search_read(
OdooModel.EXPENSE,
domain,
[
"name",
"date",
"total_amount",
"state",
"product_id",
"payment_mode",
"currency_id",
],
limit=limit,
offset=offset,
order="date desc",
)
if not expenses:
return "No expenses found."
builder = MarkdownBuilder("Expenses")
rows = []
total = 0.0
for exp in expenses:
amount = exp.get("total_amount", 0)
total += amount
currency = extract_name(exp.get("currency_id"), "EUR")
payment = "Company" if exp.get("payment_mode") == PaymentMode.COMPANY else "Employee"
rows.append([
exp["id"],
exp["date"],
exp["name"][:30],
extract_name(exp.get("product_id"), "-"),
format_money(amount, currency),
format_state(exp.get("state", ""), EXPENSE_STATE_LABELS),
payment,
])
builder.add_table(
["ID", "Date", "Name", "Category", "Amount", "Status", "Paid By"],
rows,
alignments=["right", "center", "left", "left", "right", "center", "center"],
)
builder.add_line()
builder.add_text(f"**Total: {format_money(total)}**")
builder.add_pagination(len(expenses), limit, offset)
return builder.build()
@mcp.tool()
@handle_odoo_errors
def create_expense(
name: str,
product_id: int,
total_amount: float,
description: str | None = None,
date_expense: str | None = None,
quantity: float = 1.0,
) -> str:
"""
Create a new expense report.
Args:
name: Expense name/title
product_id: Expense category ID (see list_expense_categories)
total_amount: Total expense amount
description: Detailed description (optional)
date_expense: Expense date in YYYY-MM-DD format (default: today)
quantity: Quantity (default: 1.0)
Returns:
Confirmation with created expense ID
"""
total_amount = validate_positive_amount(total_amount, "total_amount")
expense_date = (
validate_date(date_expense, "date_expense")
if date_expense
else today()
)
client = get_odoo_client()
values = {
"name": name,
"product_id": product_id,
"total_amount": total_amount,
"date": expense_date,
"quantity": quantity,
}
if description:
values["description"] = description
record_id = client.create(OdooModel.EXPENSE, values)
return f"Expense created successfully (ID: {record_id})"
@mcp.tool()
@handle_odoo_errors
def update_expense(
expense_id: int,
name: str | None = None,
total_amount: float | None = None,
description: str | None = None,
date_expense: str | None = None,
payment_mode: str | None = None,
currency: str | None = None,
analytic_account_id: int | None = None,
) -> str:
"""
Update an existing expense (only if in draft state).
Args:
expense_id: ID of expense to update
name: New name (optional)
total_amount: New amount (optional)
description: New description (optional)
date_expense: New date in YYYY-MM-DD format (optional)
payment_mode: Payment mode - "company" (paid by company) or "employee" (paid by employee) (optional)
currency: ISO currency code (e.g., "EUR", "USD") (optional)
analytic_account_id: Analytic account ID for cost distribution (optional)
Returns:
Update confirmation
"""
client = get_odoo_client()
# Check expense state
expense = client.get_record(OdooModel.EXPENSE, expense_id, ["state"])
if expense.get("state") != ExpenseState.DRAFT:
raise OdooValidationError(
"Can only update expenses in draft state",
field="state",
)
values = {}
if name is not None:
values["name"] = name
if total_amount is not None:
values["total_amount"] = validate_positive_amount(total_amount, "total_amount")
if description is not None:
values["description"] = description
if date_expense is not None:
values["date"] = validate_date(date_expense, "date_expense")
if payment_mode is not None:
mode_map = {
"company": PaymentMode.COMPANY,
"employee": PaymentMode.EMPLOYEE,
}
if payment_mode.lower() not in mode_map:
raise OdooValidationError(
f"Invalid payment_mode: {payment_mode}. Use 'company' or 'employee'",
field="payment_mode",
)
values["payment_mode"] = mode_map[payment_mode.lower()]
if currency is not None:
# Find currency ID
currencies = client.search_read(
OdooModel.CURRENCY,
[("name", "=", currency.upper())],
["id"],
limit=1,
)
if not currencies:
raise OdooValidationError(
f"Currency not found: {currency}",
field="currency",
)
values["currency_id"] = currencies[0]["id"]
if analytic_account_id is not None:
# Set analytic distribution
values["analytic_distribution"] = {str(analytic_account_id): 100}
if not values:
return "No fields to update specified."
client.write(OdooModel.EXPENSE, [expense_id], values)
return f"Expense {expense_id} updated successfully."
@mcp.tool()
@handle_odoo_errors
def delete_expense(expense_id: int) -> str:
"""
Delete an expense (only if in draft state).
Args:
expense_id: ID of expense to delete
Returns:
Deletion confirmation
"""
client = get_odoo_client()
# Check expense state
expense = client.get_record(OdooModel.EXPENSE, expense_id, ["state"])
if expense.get("state") != ExpenseState.DRAFT:
raise OdooValidationError(
"Can only delete expenses in draft state",
field="state",
)
client.unlink(OdooModel.EXPENSE, [expense_id])
return f"Expense {expense_id} deleted successfully."
@mcp.tool()
@handle_odoo_errors
def add_expense_attachment(
expense_id: int,
file_path: str,
filename: str | None = None,
) -> str:
"""
Add an attachment to an existing expense.
Args:
expense_id: Expense ID
file_path: Absolute path to file to attach
filename: Filename in Odoo (optional, uses original filename by default)
Returns:
Confirmation with created attachment ID
"""
client = get_odoo_client()
path = Path(file_path)
if not path.exists():
raise OdooValidationError(
f"File not found: {file_path}",
field="file_path",
)
# Read and encode file
with open(path, "rb") as f:
file_content = base64.b64encode(f.read()).decode("utf-8")
attachment_name = filename or path.name
attachment_id = client.create(
OdooModel.ATTACHMENT,
{
"name": attachment_name,
"type": "binary",
"datas": file_content,
"res_model": OdooModel.EXPENSE,
"res_id": expense_id,
},
)
return f"Attachment added successfully (ID: {attachment_id})"
@mcp.tool()
@handle_odoo_errors
def list_expense_attachments(expense_id: int) -> str:
"""
List attachments for an expense.
Args:
expense_id: Expense ID
Returns:
List of attachments with their ID and name
"""
client = get_odoo_client()
attachments = client.search_read(
OdooModel.ATTACHMENT,
[
("res_model", "=", OdooModel.EXPENSE),
("res_id", "=", expense_id),
],
["name", "mimetype", "file_size", "create_date"],
order="create_date desc",
)
if not attachments:
return f"No attachments found for expense {expense_id}."
builder = MarkdownBuilder(f"Attachments for Expense {expense_id}")
rows = []
for att in attachments:
size_kb = (att.get("file_size", 0) or 0) / 1024
rows.append([
att["id"],
att["name"],
att.get("mimetype", "-"),
f"{size_kb:.1f} KB",
])
builder.add_table(
["ID", "Name", "Type", "Size"],
rows,
alignments=["right", "left", "left", "right"],
)
return builder.build()
# =========================================================================
# Expense Reports (hr.expense.sheet)
# =========================================================================
@mcp.tool()
@handle_odoo_errors
def list_expense_reports(
state: str | None = None,
limit: int = 50,
offset: int = 0,
) -> str:
"""
List expense reports (expense sheets).
Args:
state: Filter by state: draft, submit, approve, post, done, cancel (optional)
limit: Maximum number of reports to return (default: 50)
offset: Offset for pagination (default: 0)
Returns:
List of expense reports with their details
"""
limit, offset = normalize_pagination(limit, offset)
client = get_odoo_client()
domain: list = []
valid_states = {"draft", "submit", "approve", "post", "done", "cancel"}
if state:
if state not in valid_states:
raise OdooValidationError(
f"Invalid state: {state}. Valid: {', '.join(valid_states)}",
field="state",
)
domain.append(("state", "=", state))
reports = client.search_read(
OdooModel.EXPENSE_SHEET,
domain,
[
"name",
"employee_id",
"total_amount",
"state",
"expense_line_ids",
"create_date",
],
limit=limit,
offset=offset,
order="create_date desc",
)
if not reports:
return "No expense reports found."
state_labels = {
"draft": "Draft",
"submit": "Submitted",
"approve": "Approved",
"post": "Posted",
"done": "Paid",
"cancel": "Cancelled",
}
builder = MarkdownBuilder("Expense Reports")
rows = []
for report in reports:
expense_count = len(report.get("expense_line_ids", []))
rows.append([
report["id"],
report.get("name", "-"),
extract_name(report.get("employee_id"), "-"),
format_money(report.get("total_amount", 0)),
expense_count,
state_labels.get(report.get("state", ""), report.get("state", "-")),
])
builder.add_table(
["ID", "Name", "Employee", "Amount", "Expenses", "Status"],
rows,
alignments=["right", "left", "left", "right", "right", "center"],
)
builder.add_pagination(len(reports), limit, offset)
return builder.build()
@mcp.tool()
@handle_odoo_errors
def create_expense_report(
name: str,
expense_ids: list[int],
) -> str:
"""
Create an expense report from selected expenses.
Args:
name: Report name/title
expense_ids: List of expense IDs to include in the report
Returns:
Confirmation with created report ID
"""
if not expense_ids:
raise OdooValidationError(
"At least one expense ID is required",
field="expense_ids",
)
client = get_odoo_client()
# Verify all expenses exist and are in draft state
expenses = client.search_read(
OdooModel.EXPENSE,
[("id", "in", expense_ids)],
["id", "state", "employee_id"],
)
if len(expenses) != len(expense_ids):
found_ids = {e["id"] for e in expenses}
missing = set(expense_ids) - found_ids
raise OdooValidationError(
f"Expenses not found: {missing}",
field="expense_ids",
)
# Check all expenses are in draft state
non_draft = [e["id"] for e in expenses if e.get("state") != ExpenseState.DRAFT]
if non_draft:
raise OdooValidationError(
f"Expenses must be in draft state. Non-draft: {non_draft}",
field="expense_ids",
)
# Check all expenses belong to the same employee
employees = {extract_name(e.get("employee_id")) for e in expenses}
if len(employees) > 1:
raise OdooValidationError(
f"All expenses must belong to the same employee. Found: {employees}",
field="expense_ids",
)
# Get the employee_id from first expense
employee_id = expenses[0].get("employee_id")
if isinstance(employee_id, (list, tuple)):
employee_id = employee_id[0]
# Create the expense report
report_id = client.create(
OdooModel.EXPENSE_SHEET,
{
"name": name,
"employee_id": employee_id,
"expense_line_ids": [(6, 0, expense_ids)],
},
)
return f"Expense report created successfully (ID: {report_id}) with {len(expense_ids)} expenses."
@mcp.tool()
@handle_odoo_errors
def submit_expense_report(report_id: int) -> str:
"""
Submit an expense report for approval.
Args:
report_id: ID of the expense report to submit
Returns:
Confirmation of submission
"""
client = get_odoo_client()
# Check report exists and is in draft state
report = client.get_record(
OdooModel.EXPENSE_SHEET,
report_id,
["state", "name"],
)
if report.get("state") != "draft":
raise OdooValidationError(
f"Report must be in draft state to submit. Current state: {report.get('state')}",
field="state",
)
# Call the action_submit_sheet method
client.execute(
OdooModel.EXPENSE_SHEET,
"action_submit_sheet",
[report_id],
)
return f"Expense report '{report.get('name')}' (ID: {report_id}) submitted for approval."
