Jarca
0f881b245f
approve dcy-001 and add REQ/BAT evidence pass selected DB path through Streamlit rule actions expose delete_category_rule MCP tool add category rule deletion regression
222 lines
13 KiB
Python
222 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from dotenv import load_dotenv
|
|
from mcp.server.fastmcp import FastMCP
|
|
|
|
from . import service
|
|
|
|
load_dotenv(Path(__file__).resolve().parent.parent / ".env")
|
|
mcp = FastMCP("jarca-ledger-mcp")
|
|
|
|
@mcp.tool()
|
|
def ledger_health_check() -> dict:
|
|
"""Verify the Jarca Ledger MCP server and SQLite database are reachable."""
|
|
return service.health_check()
|
|
|
|
@mcp.tool()
|
|
def create_ledger_account(name: str, type: str = "bank", institution: str = "", currency: str = "KRW", active: bool = True, memo: str = "", account_number_masked: str = "", holder_name: str = "", last_balance: int | None = None, last_synced_at: str = "") -> dict:
|
|
"""Create an account/card/cash/investment/loan ledger account."""
|
|
return service.create_ledger_account(name, type, institution, currency, active, memo, account_number_masked, holder_name, last_balance, last_synced_at)
|
|
|
|
@mcp.tool()
|
|
def list_ledger_accounts(active_only: bool = True) -> dict:
|
|
"""List ledger accounts and cards."""
|
|
return service.list_ledger_accounts(active_only)
|
|
|
|
@mcp.tool()
|
|
def create_ledger_transaction(transaction_date: str = "", type: str = "expense", amount: int = 0, category: str = "", subcategory: str = "", merchant: str = "", description: str = "", account_id: str = "", payment_method: str = "", person: str = "", project: str = "", tags: list[str] | str | None = None, is_fixed: bool = False, currency: str = "KRW", posted_date: str = "", source: str = "manual", external_id: str = "", memo: str = "", dedupe: bool = True) -> dict:
|
|
"""Create an income, expense, transfer, refund, or adjustment transaction. Amount is positive KRW minor unit; type determines direction."""
|
|
return service.create_ledger_transaction(transaction_date, type, amount, category, subcategory, merchant, description, account_id, payment_method, person, project, tags, is_fixed, currency, posted_date, source, external_id, memo, dedupe)
|
|
|
|
@mcp.tool()
|
|
def list_ledger_transactions(start_date: str = "", end_date: str = "", type: str = "", category: str = "", person: str = "", account_id: str = "", include_deleted: bool = False, limit: int = 100) -> dict:
|
|
"""List ledger transactions with date/category/person/account filters."""
|
|
return service.list_ledger_transactions(start_date, end_date, type, category, person, account_id, include_deleted, limit)
|
|
|
|
@mcp.tool()
|
|
def update_ledger_transaction(transaction_id: str, transaction_date: str | None = None, type: str | None = None, amount: int | None = None, category: str | None = None, subcategory: str | None = None, merchant: str | None = None, description: str | None = None, account_id: str | None = None, payment_method: str | None = None, person: str | None = None, project: str | None = None, tags: list[str] | str | None = None, is_fixed: bool | None = None, memo: str | None = None, status: str | None = None) -> dict:
|
|
"""Update selected fields on a ledger transaction."""
|
|
return service.update_ledger_transaction(transaction_id, transaction_date=transaction_date, type=type, amount=amount, category=category, subcategory=subcategory, merchant=merchant, description=description, account_id=account_id, payment_method=payment_method, person=person, project=project, tags=tags, is_fixed=is_fixed, memo=memo, status=status)
|
|
|
|
@mcp.tool()
|
|
def delete_ledger_transaction(transaction_id: str, hard_delete: bool = False) -> dict:
|
|
"""Soft-delete a ledger transaction by default, or permanently delete with hard_delete=true."""
|
|
return service.delete_ledger_transaction(transaction_id, hard_delete)
|
|
|
|
@mcp.tool()
|
|
def create_ledger_budget(month: str, category: str, amount: int, subcategory: str = "", person: str = "", currency: str = "KRW", memo: str = "") -> dict:
|
|
"""Create or update a monthly category/person budget."""
|
|
return service.create_ledger_budget(month, category, amount, subcategory, person, currency, memo)
|
|
|
|
@mcp.tool()
|
|
def list_ledger_budgets(month: str = "") -> dict:
|
|
"""List ledger budgets, optionally for a month YYYY-MM."""
|
|
return service.list_ledger_budgets(month)
|
|
|
|
@mcp.tool()
|
|
def get_monthly_cashflow_report(month: str = "") -> dict:
|
|
"""Return income, expenses, net cashflow, and savings rate for a month."""
|
|
return service.get_monthly_cashflow_report(month)
|
|
|
|
@mcp.tool()
|
|
def get_category_spending_report(month: str = "", start_date: str = "", end_date: str = "", person: str = "") -> dict:
|
|
"""Return category/subcategory spending for a month or date range."""
|
|
return service.get_category_spending_report(month, start_date, end_date, person)
|
|
|
|
@mcp.tool()
|
|
def get_fixed_cost_report(month: str = "") -> dict:
|
|
"""Return fixed expense transactions and total fixed cost for a month."""
|
|
return service.get_fixed_cost_report(month)
|
|
|
|
@mcp.tool()
|
|
def get_budget_vs_actual_report(month: str = "") -> dict:
|
|
"""Compare monthly budgets against actual spending."""
|
|
return service.get_budget_vs_actual_report(month)
|
|
|
|
@mcp.tool()
|
|
def get_investable_cashflow_report(month: str = "", minimum_cash_buffer: int = 0, planned_extra_expense: int = 0) -> dict:
|
|
"""Estimate cashflow available for investing after buffer and planned extra expenses."""
|
|
return service.get_investable_cashflow_report(month, minimum_cash_buffer, planned_extra_expense)
|
|
|
|
@mcp.tool()
|
|
def preview_ledger_csv_import(file_path: str, source: str = "generic") -> dict:
|
|
"""Preview a UTF-8/UTF-8-BOM CSV import before saving transactions."""
|
|
return service.preview_ledger_csv_import(file_path, source)
|
|
|
|
@mcp.tool()
|
|
def preview_kb_bank_pdf_import(file_path: str, password: str, preview_limit: int = 20) -> dict:
|
|
"""Preview transactions from a password-protected KB Kookmin Bank account-history PDF without saving them."""
|
|
return service.preview_kb_bank_pdf_import(file_path, password, preview_limit)
|
|
|
|
@mcp.tool()
|
|
def import_kb_bank_pdf(file_path: str, password: str, account_id: str = "", skip_duplicates: bool = True, skip_validation: bool = False) -> dict:
|
|
"""Import transactions from a KB Kookmin Bank account-history PDF into the ledger database."""
|
|
return service.import_kb_bank_pdf(file_path, password, account_id, skip_duplicates, skip_validation=skip_validation)
|
|
|
|
@mcp.tool()
|
|
def preview_seoulpay_xlsx_import(file_path: str, password: str = "", preview_limit: int = 20) -> dict:
|
|
"""Preview transactions from a SeoulPay usage-history XLSX before saving them."""
|
|
return service.preview_seoulpay_xlsx_import(file_path, password, preview_limit)
|
|
|
|
@mcp.tool()
|
|
def import_seoulpay_xlsx(file_path: str, password: str = "", account_id: str = "", skip_duplicates: bool = True) -> dict:
|
|
"""Import transactions from a SeoulPay usage-history XLSX into the ledger database."""
|
|
return service.import_seoulpay_xlsx(file_path, password, account_id, skip_duplicates)
|
|
|
|
@mcp.tool()
|
|
def preview_toss_securities_pdf_import(file_path: str, preview_limit: int = 20) -> dict:
|
|
"""Preview transactions from a Toss Securities transaction-history PDF without saving them."""
|
|
return service.preview_toss_securities_pdf_import(file_path, preview_limit)
|
|
|
|
@mcp.tool()
|
|
def import_toss_securities_pdf(file_path: str, account_id: str = "", skip_duplicates: bool = True) -> dict:
|
|
"""Import transactions from a Toss Securities transaction-history PDF into the ledger database."""
|
|
return service.import_toss_securities_pdf(file_path, account_id, skip_duplicates)
|
|
|
|
@mcp.tool()
|
|
def preview_toss_bank_pdf_import(file_path: str, preview_limit: int = 20) -> dict:
|
|
"""Preview transactions from a Toss Bank transaction-history PDF without saving them."""
|
|
return service.preview_toss_bank_pdf_import(file_path, preview_limit)
|
|
|
|
@mcp.tool()
|
|
def import_toss_bank_pdf(file_path: str, account_id: str = "", skip_duplicates: bool = True) -> dict:
|
|
"""Import transactions from a Toss Bank transaction-history PDF into the ledger database."""
|
|
return service.import_toss_bank_pdf(file_path, account_id, skip_duplicates)
|
|
|
|
@mcp.tool()
|
|
def list_ledger_import_batches(source: str = "", include_rolled_back: bool = True, limit: int = 50) -> dict:
|
|
"""List CSV/PDF/API import batches."""
|
|
return service.list_ledger_import_batches(source, include_rolled_back, limit)
|
|
|
|
@mcp.tool()
|
|
def rollback_ledger_import_batch(import_batch_id: str, hard_delete: bool = False) -> dict:
|
|
"""Roll back all transactions created by one import batch. Soft-deletes by default."""
|
|
return service.rollback_ledger_import_batch(import_batch_id, hard_delete)
|
|
|
|
@mcp.tool()
|
|
def create_category_rule(match_field: str, pattern: str, category: str, subcategory: str = "", transaction_type: str = "", match_type: str = "contains", priority: int = 100, person: str = "", is_fixed: bool | None = None, confidence: float = 1.0, memo: str = "", amount_min: int | None = None, amount_max: int | None = None, day_of_month: int | None = None) -> dict:
|
|
"""Create or update a ledger transaction categorization rule."""
|
|
return service.create_category_rule(match_field, pattern, category, subcategory, transaction_type, match_type, priority, person, is_fixed, confidence, memo, amount_min, amount_max, day_of_month)
|
|
|
|
@mcp.tool()
|
|
def seed_default_category_rules(overwrite: bool = False) -> dict:
|
|
"""Seed default Jarca category rules for common Korean household ledger patterns."""
|
|
return service.seed_default_category_rules(overwrite)
|
|
|
|
@mcp.tool()
|
|
def list_category_rules(active_only: bool = True) -> dict:
|
|
"""List transaction categorization rules."""
|
|
return service.list_category_rules(active_only)
|
|
|
|
@mcp.tool()
|
|
def update_category_rule(rule_id: str, match_field: str, pattern: str, category: str, subcategory: str = "", transaction_type: str = "", match_type: str = "contains", priority: int = 100, person: str = "", is_fixed: bool | None = None, confidence: float = 1.0, memo: str = "", amount_min: int | None = None, amount_max: int | None = None, day_of_month: int | None = None, active: bool = True) -> dict:
|
|
"""Update an existing ledger transaction categorization rule."""
|
|
return service.update_category_rule(rule_id, match_field, pattern, category, subcategory, transaction_type, match_type, priority, person, is_fixed, confidence, memo, amount_min, amount_max, day_of_month, active)
|
|
|
|
@mcp.tool()
|
|
def delete_category_rule(rule_id: str) -> dict:
|
|
"""Delete a transaction categorization rule."""
|
|
return service.delete_category_rule(rule_id)
|
|
|
|
@mcp.tool()
|
|
def preview_categorization(limit: int = 1000, only_uncategorized: bool = True) -> dict:
|
|
"""Preview category-rule matches before writing categories to transactions."""
|
|
return service.preview_categorization(limit, only_uncategorized)
|
|
|
|
@mcp.tool()
|
|
def apply_categorization_rules(limit: int = 1000, only_uncategorized: bool = True) -> dict:
|
|
"""Apply active category rules to matching ledger transactions."""
|
|
return service.apply_categorization_rules(limit, only_uncategorized)
|
|
|
|
@mcp.tool()
|
|
def list_uncategorized_transactions(limit: int = 100, min_amount: int = 0) -> dict:
|
|
"""List active transactions that still have no category."""
|
|
return service.list_uncategorized_transactions(limit, min_amount)
|
|
|
|
@mcp.tool()
|
|
def get_uncategorized_merchant_summary(limit: int = 50) -> dict:
|
|
"""Summarize uncategorized transactions by merchant/raw_summary/type for user review."""
|
|
return service.get_uncategorized_merchant_summary(limit)
|
|
|
|
@mcp.tool()
|
|
def create_ledger_category(category: str, subcategory: str = "", sort_order: int = 100, memo: str = "", cashflow_kind: str = "") -> dict:
|
|
"""Create or reactivate a ledger category/subcategory used by the dashboard and transactions."""
|
|
return service.create_ledger_category(category, subcategory, sort_order, memo, cashflow_kind)
|
|
|
|
@mcp.tool()
|
|
def seed_default_ledger_categories(overwrite: bool = False) -> dict:
|
|
"""Seed default ledger categories and subcategories."""
|
|
return service.seed_default_ledger_categories(overwrite)
|
|
|
|
@mcp.tool()
|
|
def list_ledger_categories(active_only: bool = True, include_usage: bool = True) -> dict:
|
|
"""List ledger categories and subcategories, optionally with usage counts."""
|
|
return service.list_ledger_categories(active_only, include_usage)
|
|
|
|
@mcp.tool()
|
|
def list_cross_transfer_review_queue(limit: int = 300) -> dict:
|
|
"""List KB↔토스 계좌 이체 확인필요 큐 항목."""
|
|
return service.list_cross_transfer_review_queue(limit)
|
|
|
|
@mcp.tool()
|
|
def preview_cross_account_transfers(limit: int = 500) -> dict:
|
|
"""Preview KB↔토스은행 이체 매칭 후보를 반환합니다."""
|
|
return service.preview_cross_account_transfers(limit)
|
|
|
|
@mcp.tool()
|
|
def run_cross_account_transfer_match(limit: int = 500) -> dict:
|
|
"""Run KB↔토스은행 자동 이체 매칭 및 태그/분류 반영을 수행합니다."""
|
|
return service.run_cross_account_transfer_match(limit)
|
|
|
|
@mcp.tool()
|
|
def delete_ledger_category(category: str, subcategory: str = "", hard_delete: bool = False, force: bool = False) -> dict:
|
|
"""Deactivate or delete a ledger category/subcategory."""
|
|
return service.delete_ledger_category(category, subcategory, hard_delete, force)
|
|
|
|
def main() -> None:
|
|
mcp.run()
|
|
|
|
if __name__ == "__main__":
|
|
main()
|