"""MCP Tool implementations."""
import json
from typing import Any
from pydantic import ValidationError
from .errors import OracleApiError
from .formatter import (
create_invoice_summary,
format_invoice_details_markdown,
format_invoice_list_markdown,
format_search_results_markdown,
format_error_message,
)
from .models import (
GetInvoiceDetailsParams,
ListInvoicesParams,
SearchInvoicesParams,
)
from .oracle_client import oracle_client
async def list_invoices(arguments: dict[str, Any]) -> list[dict[str, Any]]:
"""Tool: oracle_ar_list_invoices - List and filter invoices."""
try:
# Validate input
params = ListInvoicesParams(**arguments)
# Call Oracle API
api_response = await oracle_client.list_invoices(
username=params.username,
password=params.password,
limit=params.limit,
offset=params.offset,
customer_name=params.customer_name,
invoice_number=params.invoice_number,
date_from=params.date_from,
date_to=params.date_to,
status=params.status,
)
# Create summary with totals
summary = create_invoice_summary(api_response)
# Format response
markdown = format_invoice_list_markdown(summary)
json_data = summary.model_dump_json(indent=2)
return [
{"type": "text", "text": markdown},
{"type": "text", "text": f"\n\n---\n\n**Raw JSON Data:**\n```json\n{json_data}\n```"},
]
except ValidationError as e:
error_msg = format_error_message(e)
return [{"type": "text", "text": error_msg}]
except OracleApiError as e:
error_msg = format_error_message(e)
return [{"type": "text", "text": error_msg}]
except Exception as e:
error_msg = format_error_message(e)
return [{"type": "text", "text": error_msg}]
async def get_invoice_details(arguments: dict[str, Any]) -> list[dict[str, Any]]:
"""Tool: oracle_ar_get_invoice_details - Get specific invoice details."""
try:
# Validate input
params = GetInvoiceDetailsParams(**arguments)
# Call Oracle API
invoice = await oracle_client.get_invoice_details(
username=params.username,
password=params.password,
invoice_id=params.invoice_id,
)
# Format response
markdown = format_invoice_details_markdown(invoice)
json_data = invoice.model_dump_json(indent=2)
return [
{"type": "text", "text": markdown},
{"type": "text", "text": f"\n\n---\n\n**Raw JSON Data:**\n```json\n{json_data}\n```"},
]
except ValidationError as e:
error_msg = format_error_message(e)
return [{"type": "text", "text": error_msg}]
except OracleApiError as e:
error_msg = format_error_message(e)
return [{"type": "text", "text": error_msg}]
except Exception as e:
error_msg = format_error_message(e)
return [{"type": "text", "text": error_msg}]
async def search_invoices(arguments: dict[str, Any]) -> list[dict[str, Any]]:
"""Tool: oracle_ar_search_invoices - Advanced search with filters."""
try:
# Validate input
params = SearchInvoicesParams(**arguments)
# Call Oracle API
api_response = await oracle_client.search_invoices(
username=params.username,
password=params.password,
q=params.q,
filters=params.filters,
limit=params.limit,
offset=params.offset,
)
# Create summary with totals
summary = create_invoice_summary(api_response)
# Calculate additional statistics
open_invoices = [inv for inv in summary.items if inv.InvoiceStatus == "Open"]
overdueInvoices = [
inv
for inv in open_invoices
if get_days_overdue(inv.DueDate) > 0
]
# Format response
markdown = format_search_results_markdown(summary, params.q)
# Add statistics
markdown += "\n## Additional Statistics\n\n"
markdown += f"- **Open Invoices**: {len(open_invoices)}\n"
markdown += f"- **Overdue Invoices**: {len(overdueInvoices)}\n"
if params.filters:
markdown += "\n## Applied Filters\n\n"
if params.filters.amountGreaterThan is not None:
markdown += f"- Amount greater than: ${params.filters.amountGreaterThan}\n"
if params.filters.amountLessThan is not None:
markdown += f"- Amount less than: ${params.filters.amountLessThan}\n"
if params.filters.overdueDays is not None:
markdown += f"- Overdue by: {params.filters.overdueDays}+ days\n"
if params.filters.customerId:
markdown += f"- Customer ID: {params.filters.customerId}\n"
json_data = summary.model_dump_json(indent=2)
return [
{"type": "text", "text": markdown},
{"type": "text", "text": f"\n\n---\n\n**Raw JSON Data:**\n```json\n{json_data}\n```"},
]
except ValidationError as e:
error_msg = format_error_message(e)
return [{"type": "text", "text": error_msg}]
except OracleApiError as e:
error_msg = format_error_message(e)
return [{"type": "text", "text": error_msg}]
except Exception as e:
error_msg = format_error_message(e)
return [{"type": "text", "text": error_msg}]
# Import for get_days_overdue
from .formatter import get_days_overdue
