Accounting MCP Server

""" MCP 工具实现 - 记账相关的 MCP Tools 实现所有记账相关的 MCP 工具,需要先执行conda activate myenv，才能使用下列工具： - add_transaction: 添加交易记录 - get_balance: 获取账户余额 - list_transactions: 查询交易记录 - get_monthly_summary: 获取月度汇总 """ from datetime import date, datetime from decimal import Decimal from typing import Dict, Any, List, Optional, Union import json from .models import Transaction, TransactionType from .storage import StorageManager class AccountingTools: """记账工具类 - 实现所有 MCP Tools""" def __init__(self, storage_manager: Optional[StorageManager] = None): self.storage = storage_manager or StorageManager() async def add_transaction( self, amount: Union[float, int, str], category: str, description: str = "", date_str: Optional[str] = None ) -> Dict[str, Any]: """ 添加交易记录 Args: amount: 金额，正数为收入，负数为支出 category: 分类ID description: 交易描述 date_str: 交易日期 (YYYY-MM-DD 格式，可选) Returns: Dict containing transaction details and updated balance """ try: # 转换金额为 Decimal amount_decimal = Decimal(str(amount)) # 解析日期 transaction_date = date.today() if date_str: transaction_date = datetime.fromisoformat(date_str).date() # 创建交易记录 transaction = Transaction( amount=amount_decimal, category=category, description=description, date=transaction_date ) # 添加交易到存储 self.storage.add_transaction(transaction) # 获取更新后的余额 new_balance = self.storage.get_balance() return { "success": True, "transaction": { "id": transaction.id, "amount": float(transaction.amount), "category": transaction.category, "description": transaction.description, "date": transaction.date.isoformat(), "type": transaction.transaction_type.value }, "new_balance": float(new_balance), "message": f"已添加{transaction.transaction_type.value == 'income' and '收入' or '支出'} ¥{abs(float(transaction.amount))}" } except ValueError as e: return { "success": False, "error": f"参数错误: {str(e)}", "error_type": "validation_error" } except Exception as e: return { "success": False, "error": f"添加交易失败: {str(e)}", "error_type": "storage_error" } async def get_balance(self, detailed: bool = False) -> Dict[str, Any]: """ 获取账户余额信息 Args: detailed: 是否返回详细统计信息 Returns: Dict containing balance and optionally detailed statistics """ try: account = self.storage.account.load_account() result = { "success": True, "balance": float(account.balance), "total_transactions": account.total_transactions, "last_updated": account.last_updated.isoformat() if account.last_updated else None } if detailed: # 获取本月交易统计 now = datetime.now() monthly_summary = self.storage.get_monthly_summary(now.year, now.month) result["monthly_stats"] = { "income": monthly_summary["total_income"], "expense": monthly_summary["total_expense"], "net_flow": monthly_summary["net_flow"], "transaction_count": monthly_summary["transaction_count"] } return result except Exception as e: return { "success": False, "error": f"获取余额失败: {str(e)}", "error_type": "storage_error" } async def list_transactions( self, limit: int = 20, offset: int = 0, category: Optional[str] = None, start_date: Optional[str] = None, end_date: Optional[str] = None ) -> Dict[str, Any]: """ 查询交易记录列表 Args: limit: 返回记录数限制 offset: 偏移量（用于分页） category: 分类筛选 start_date: 开始日期 (YYYY-MM-DD) end_date: 结束日期 (YYYY-MM-DD) Returns: Dict containing transaction list and metadata """ try: # 解析日期筛选条件 start_date_obj = None end_date_obj = None if start_date: start_date_obj = datetime.fromisoformat(start_date).date() if end_date: end_date_obj = datetime.fromisoformat(end_date).date() # 获取交易记录 transactions = self.storage.transactions.get_transactions( limit=limit, offset=offset, category=category, start_date=start_date_obj, end_date=end_date_obj ) # 转换为字典格式 transaction_list = [ { "id": t.id, "amount": float(t.amount), "category": t.category, "description": t.description, "date": t.date.isoformat(), "timestamp": t.timestamp.isoformat(), "type": t.transaction_type.value } for t in transactions ] # 获取总数（用于分页信息） all_transactions = self.storage.transactions.load_transactions() total_count = len(all_transactions) return { "success": True, "transactions": transaction_list, "pagination": { "limit": limit, "offset": offset, "total_count": total_count, "returned_count": len(transaction_list), "has_more": offset + len(transaction_list) < total_count }, "filters": { "category": category, "start_date": start_date, "end_date": end_date } } except Exception as e: return { "success": False, "error": f"查询交易记录失败: {str(e)}", "error_type": "storage_error" } async def get_monthly_summary( self, year: Optional[int] = None, month: Optional[int] = None ) -> Dict[str, Any]: """ 获取月度财务汇总 Args: year: 年份（默认当前年） month: 月份（默认当前月） Returns: Dict containing monthly financial summary """ try: # 默认使用当前年月 now = datetime.now() year = year or now.year month = month or now.month # 验证日期有效性 if not (1 <= month <= 12): raise ValueError("月份必须在1-12之间") if year < 1900 or year > 2100: raise ValueError("年份必须在1900-2100之间") # 获取月度汇总 summary = self.storage.get_monthly_summary(year, month) # 获取分类信息 categories = self.storage.categories.load_categories() category_info = {cat.id: {"name": cat.name, "icon": cat.icon} for cat in categories} # 增强分类统计信息 enhanced_breakdown = {} for cat_name, amount in summary["category_breakdown"].items(): # 查找对应的分类信息 cat_info = None for cat in categories: if cat.name == cat_name: cat_info = cat break enhanced_breakdown[cat_name] = { "amount": amount, "icon": cat_info.icon if cat_info else "📝", "id": cat_info.id if cat_info else "unknown" } return { "success": True, "summary": { "period": { "year": year, "month": month, "month_name": datetime(year, month, 1).strftime("%B") }, "totals": { "income": summary["total_income"], "expense": summary["total_expense"], "net_flow": summary["net_flow"] }, "transaction_count": summary["transaction_count"], "category_breakdown": enhanced_breakdown, "top_categories": sorted( enhanced_breakdown.items(), key=lambda x: x[1]["amount"], reverse=True )[:5] # 前5大支出分类 } } except ValueError as e: return { "success": False, "error": f"参数错误: {str(e)}", "error_type": "validation_error" } except Exception as e: return { "success": False, "error": f"获取月度汇总失败: {str(e)}", "error_type": "storage_error" } async def get_categories(self) -> Dict[str, Any]: """ 获取所有分类列表（额外工具） Returns: Dict containing all available categories """ try: categories = self.storage.categories.load_categories() category_list = [ { "id": cat.id, "name": cat.name, "description": cat.description, "type": cat.category_type.value, "color": cat.color, "icon": cat.icon } for cat in categories ] # 按类型分组 expense_categories = [cat for cat in category_list if cat["type"] == "expense"] income_categories = [cat for cat in category_list if cat["type"] == "income"] return { "success": True, "categories": { "all": category_list, "expense": expense_categories, "income": income_categories }, "total_count": len(category_list) } except Exception as e: return { "success": False, "error": f"获取分类列表失败: {str(e)}", "error_type": "storage_error" } # 工具函数 - 用于 MCP 服务器注册 def get_tool_definitions() -> List[Dict[str, Any]]: """返回所有工具的 MCP 定义""" return [ { "name": "add_transaction", "description": "添加一笔收入或支出记录到账本中", "readOnly": False, "destructive": False, "inputSchema": { "type": "object", "properties": { "amount": { "type": "number", "description": "金额，正数为收入，负数为支出" }, "category": { "type": "string", "description": "分类：food(餐饮), transport(交通), entertainment(娱乐), shopping(购物), healthcare(医疗), education(教育), income(收入), other(其他)" }, "description": { "type": "string", "description": "交易描述信息（可选）", "default": "" }, "date": { "type": "string", "description": "交易日期，格式为 YYYY-MM-DD，默认为今天", "format": "date" } }, "required": ["amount", "category"] } }, { "name": "get_balance", "description": "获取账户当前余额和基本统计信息", "readOnly": True, "destructive": False, "inputSchema": { "type": "object", "properties": { "detailed": { "type": "boolean", "description": "是否返回详细的月度统计信息", "default": False } } } }, { "name": "list_transactions", "description": "查询交易记录，支持分页和条件筛选", "readOnly": True, "destructive": False, "inputSchema": { "type": "object", "properties": { "limit": { "type": "integer", "description": "返回记录数限制，默认20条", "default": 20, "minimum": 1, "maximum": 100 }, "offset": { "type": "integer", "description": "跳过的记录数，用于分页", "default": 0, "minimum": 0 }, "category": { "type": "string", "description": "按分类筛选交易记录" }, "start_date": { "type": "string", "description": "开始日期，格式为 YYYY-MM-DD", "format": "date" }, "end_date": { "type": "string", "description": "结束日期，格式为 YYYY-MM-DD", "format": "date" } } } }, { "name": "get_monthly_summary", "description": "获取指定月份的收支汇总统计，包括分类统计", "readOnly": True, "destructive": False, "inputSchema": { "type": "object", "properties": { "year": { "type": "integer", "description": "年份，默认为当前年份", "minimum": 1900, "maximum": 2100 }, "month": { "type": "integer", "description": "月份 (1-12)，默认为当前月份", "minimum": 1, "maximum": 12 } } } }, { "name": "get_categories", "description": "获取所有可用的支出和收入分类列表", "readOnly": True, "destructive": False, "inputSchema": { "type": "object", "properties": {} } } ]