invoices.py•13.8 kB
"""Invoice management tools for B4B MCP server."""
from typing import Optional
import httpx
import structlog
from mcp.server.fastmcp import Context
from ..auth.token_handler import extract_user_token
from ..client.api_client import get_api_client
from ..server import mcp
from ..utils.validators import validate_uuid, validate_page_params
from ..utils.errors import MCPValidationError, MCPAuthenticationError, MCPAuthorizationError
logger = structlog.get_logger(__name__)
@mcp.tool()
async def list_invoices(
entity_id: str,
page: int = 1,
per_page: int = 20,
date_from: Optional[str] = None,
date_to: Optional[str] = None,
status: Optional[int] = None,
search: Optional[str] = None,
ctx: Context = None,
) -> dict:
"""
List invoices for an entity with filtering options.
Args:
entity_id: UUID of the entity
page: Page number (default: 1)
per_page: Items per page, max 100 (default: 20)
date_from: Start date filter (ISO format: YYYY-MM-DD)
date_to: End date filter (ISO format: YYYY-MM-DD)
status: Invoice status (0=DRAFT, 1=ACTIVE, 2=CANCELLED, 3=ARCHIVED)
search: Search in invoice number or vendor name
ctx: MCP context (automatically provided)
Returns:
Paginated list of invoices with vendor info
"""
log = logger.bind(action="list_invoices", entity_id=entity_id, page=page, per_page=per_page)
log.info("Starting list invoices")
try:
# Extract user token
token = await extract_user_token(ctx)
# Validate entity_id
entity_id = validate_uuid(entity_id, "entity_id")
# Validate pagination
page, per_page = validate_page_params(page, per_page)
# Build query params
params = {
"page": page,
"per_page": per_page,
}
# Add optional filters
if date_from is not None:
params["date_from"] = date_from
if date_to is not None:
params["date_to"] = date_to
if status is not None:
params["status"] = status
if search is not None:
params["search"] = search
# Get API client
api_client = get_api_client()
# Call B4B API
response_data = await api_client.get(
f"/entities/{entity_id}/invoices",
token=token,
params=params,
)
log.info(
"List invoices successful",
total_invoices=response_data.get("total", 0),
page=page,
)
return {
"success": True,
"data": response_data,
}
except MCPAuthenticationError as e:
log.warning("Authentication failed", error=str(e))
return {
"success": False,
"error": str(e),
}
except MCPAuthorizationError as e:
log.warning("Authorization failed", error=str(e))
return {
"success": False,
"error": str(e),
}
except MCPValidationError as e:
log.warning("Validation failed", error=str(e))
return {
"success": False,
"error": str(e),
}
except httpx.HTTPStatusError as e:
log.error("HTTP error listing invoices", status_code=e.response.status_code)
error_detail = ""
try:
error_body = e.response.json()
error_detail = error_body.get("detail", "") or error_body.get("message", "")
except Exception:
error_detail = e.response.text
if e.response.status_code == 403:
return {
"success": False,
"error": "Access denied. You do not have permission to access this entity's invoices.",
}
elif e.response.status_code == 404:
return {
"success": False,
"error": "Entity not found.",
}
else:
return {
"success": False,
"error": f"Failed to list invoices: {error_detail or f'HTTP {e.response.status_code}'}",
}
except Exception as e:
log.error("Unexpected error listing invoices", error=str(e))
return {
"success": False,
"error": f"Unexpected error listing invoices: {str(e)}",
}
@mcp.tool()
async def get_invoice(entity_id: str, invoice_id: str, ctx: Context) -> dict:
"""
Get detailed information about a specific invoice.
Args:
entity_id: UUID of the entity
invoice_id: UUID of the invoice
ctx: MCP context (automatically provided)
Returns:
Complete invoice details including line items
"""
log = logger.bind(action="get_invoice", entity_id=entity_id, invoice_id=invoice_id)
log.info("Starting get invoice")
try:
# Extract user token
token = await extract_user_token(ctx)
# Validate UUIDs
entity_id = validate_uuid(entity_id, "entity_id")
invoice_id = validate_uuid(invoice_id, "invoice_id")
# Get API client
api_client = get_api_client()
# Call B4B API
response_data = await api_client.get(
f"/entities/{entity_id}/invoices/{invoice_id}",
token=token,
)
log.info(
"Get invoice successful",
invoice_number=response_data.get("invoice_number", "N/A"),
)
return {
"success": True,
"data": response_data,
}
except MCPAuthenticationError as e:
log.warning("Authentication failed", error=str(e))
return {
"success": False,
"error": str(e),
}
except MCPAuthorizationError as e:
log.warning("Authorization failed", error=str(e))
return {
"success": False,
"error": str(e),
}
except MCPValidationError as e:
log.warning("Validation failed", error=str(e))
return {
"success": False,
"error": str(e),
}
except httpx.HTTPStatusError as e:
log.error("HTTP error getting invoice", status_code=e.response.status_code)
error_detail = ""
try:
error_body = e.response.json()
error_detail = error_body.get("detail", "") or error_body.get("message", "")
except Exception:
error_detail = e.response.text
if e.response.status_code == 403:
return {
"success": False,
"error": "Access denied. You do not have permission to access this invoice.",
}
elif e.response.status_code == 404:
return {
"success": False,
"error": "Invoice not found.",
}
else:
return {
"success": False,
"error": f"Failed to get invoice: {error_detail or f'HTTP {e.response.status_code}'}",
}
except Exception as e:
log.error("Unexpected error getting invoice", error=str(e))
return {
"success": False,
"error": f"Unexpected error getting invoice: {str(e)}",
}
@mcp.tool()
async def import_invoice_xml(entity_id: str, xml_content: str, ctx: Context = None) -> dict:
"""
Import invoice from Vietnamese e-invoice XML format.
Args:
entity_id: UUID of the entity
xml_content: XML content as string
ctx: MCP context (automatically provided)
Returns:
Imported invoice details with parsing status
"""
log = logger.bind(action="import_invoice_xml", entity_id=entity_id)
log.info("Starting import invoice XML")
try:
# Extract user token
token = await extract_user_token(ctx)
# Validate entity_id
entity_id = validate_uuid(entity_id, "entity_id")
# Validate XML content
if not xml_content or not xml_content.strip():
raise MCPValidationError("XML content cannot be empty", "xml_content")
# Get API client
api_client = get_api_client()
# Call B4B API
response_data = await api_client.post(
f"/entities/{entity_id}/invoices/upload-xml",
token=token,
json={"xml_content": xml_content},
)
log.info(
"Import invoice XML successful",
invoice_id=response_data.get("id", "N/A"),
)
return {
"success": True,
"data": response_data,
}
except MCPAuthenticationError as e:
log.warning("Authentication failed", error=str(e))
return {
"success": False,
"error": str(e),
}
except MCPAuthorizationError as e:
log.warning("Authorization failed", error=str(e))
return {
"success": False,
"error": str(e),
}
except MCPValidationError as e:
log.warning("Validation failed", error=str(e))
return {
"success": False,
"error": str(e),
}
except httpx.HTTPStatusError as e:
log.error("HTTP error importing invoice XML", status_code=e.response.status_code)
error_detail = ""
try:
error_body = e.response.json()
error_detail = error_body.get("detail", "") or error_body.get("message", "")
except Exception:
error_detail = e.response.text
if e.response.status_code == 400:
return {
"success": False,
"error": f"Failed to parse XML: {error_detail or 'Invalid XML format'}",
}
elif e.response.status_code == 403:
return {
"success": False,
"error": "Access denied. You do not have permission to import invoices for this entity.",
}
elif e.response.status_code == 404:
return {
"success": False,
"error": "Entity not found.",
}
else:
return {
"success": False,
"error": f"Failed to import invoice: {error_detail or f'HTTP {e.response.status_code}'}",
}
except Exception as e:
log.error("Unexpected error importing invoice XML", error=str(e))
return {
"success": False,
"error": f"Unexpected error importing invoice: {str(e)}",
}
@mcp.tool()
async def get_invoice_statistics(
entity_id: str,
year: Optional[int] = None,
month: Optional[int] = None,
ctx: Context = None,
) -> dict:
"""
Get invoice statistics and analytics.
Args:
entity_id: UUID of the entity
year: Optional year filter
month: Optional month filter (1-12)
ctx: MCP context (automatically provided)
Returns:
Statistics including total invoices, amounts, VAT, etc.
"""
log = logger.bind(action="get_invoice_statistics", entity_id=entity_id, year=year, month=month)
log.info("Starting get invoice statistics")
try:
# Extract user token
token = await extract_user_token(ctx)
# Validate entity_id
entity_id = validate_uuid(entity_id, "entity_id")
# Build query params
params = {}
if year is not None:
params["year"] = year
if month is not None:
if month < 1 or month > 12:
raise MCPValidationError("Month must be between 1 and 12", "month")
params["month"] = month
# Get API client
api_client = get_api_client()
# Call B4B API
response_data = await api_client.get(
f"/entities/{entity_id}/invoices/stats",
token=token,
params=params if params else None,
)
log.info(
"Get invoice statistics successful",
total_invoices=response_data.get("total_invoices", 0),
)
return {
"success": True,
"data": response_data,
}
except MCPAuthenticationError as e:
log.warning("Authentication failed", error=str(e))
return {
"success": False,
"error": str(e),
}
except MCPAuthorizationError as e:
log.warning("Authorization failed", error=str(e))
return {
"success": False,
"error": str(e),
}
except MCPValidationError as e:
log.warning("Validation failed", error=str(e))
return {
"success": False,
"error": str(e),
}
except httpx.HTTPStatusError as e:
log.error("HTTP error getting invoice statistics", status_code=e.response.status_code)
error_detail = ""
try:
error_body = e.response.json()
error_detail = error_body.get("detail", "") or error_body.get("message", "")
except Exception:
error_detail = e.response.text
if e.response.status_code == 403:
return {
"success": False,
"error": "Access denied. You do not have permission to access this entity's statistics.",
}
elif e.response.status_code == 404:
return {
"success": False,
"error": "Entity not found.",
}
else:
return {
"success": False,
"error": f"Failed to get statistics: {error_detail or f'HTTP {e.response.status_code}'}",
}
except Exception as e:
log.error("Unexpected error getting invoice statistics", error=str(e))
return {
"success": False,
"error": f"Unexpected error getting statistics: {str(e)}",
}