Accounting MCP Server

""" MCP 资源实现 - 记账数据资源 实现所有记账相关的 MCP Resources： - transactions://all - 所有交易记录 - categories://list - 分类列表 - summary://current - 当前汇总信息 """ from typing import Dict, Any, List, Optional import json from datetime import datetime from .storage import StorageManager class AccountingResources: """记账资源类 - 实现所有 MCP Resources""" def __init__(self, storage_manager: Optional[StorageManager] = None): self.storage = storage_manager or StorageManager() async def get_resource(self, uri: str) -> Dict[str, Any]: """ 根据 URI 获取资源内容 Args: uri: 资源 URI Returns: Dict containing resource content and metadata """ try: if uri == "transactions://all": return await self._get_all_transactions() elif uri == "categories://list": return await self._get_categories_list() elif uri == "summary://current": return await self._get_current_summary() else: return { "success": False, "error": f"未知的资源 URI: {uri}", "error_type": "invalid_uri" } except Exception as e: return { "success": False, "error": f"获取资源失败: {str(e)}", "error_type": "resource_error" } async def _get_all_transactions(self) -> Dict[str, Any]: """获取所有交易记录资源""" transactions = self.storage.transactions.load_transactions() # 转换为资源格式 transactions_data = [] for t in transactions: transactions_data.append({ "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 }) # 计算统计信息 total_income = sum(t["amount"] for t in transactions_data if t["amount"] > 0) total_expense = sum(abs(t["amount"]) for t in transactions_data if t["amount"] < 0) return { "success": True, "content": json.dumps({ "transactions": transactions_data, "metadata": { "total_count": len(transactions_data), "total_income": total_income, "total_expense": total_expense, "date_range": { "earliest": min(t["date"] for t in transactions_data) if transactions_data else None, "latest": max(t["date"] for t in transactions_data) if transactions_data else None }, "generated_at": datetime.now().isoformat() } }, indent=2, ensure_ascii=False), "mimeType": "application/json" } async def _get_categories_list(self) -> Dict[str, Any]: """获取分类列表资源""" categories = self.storage.categories.load_categories() # 转换为资源格式 categories_data = [] for cat in categories: categories_data.append({ "id": cat.id, "name": cat.name, "description": cat.description, "type": cat.category_type.value, "color": cat.color, "icon": cat.icon }) # 按类型分组 expense_categories = [cat for cat in categories_data if cat["type"] == "expense"] income_categories = [cat for cat in categories_data if cat["type"] == "income"] return { "success": True, "content": json.dumps({ "categories": { "all": categories_data, "by_type": { "expense": expense_categories, "income": income_categories } }, "metadata": { "total_count": len(categories_data), "expense_count": len(expense_categories), "income_count": len(income_categories), "generated_at": datetime.now().isoformat() } }, indent=2, ensure_ascii=False), "mimeType": "application/json" } async def _get_current_summary(self) -> Dict[str, Any]: """获取当前汇总信息资源""" # 获取账户信息 account = self.storage.account.load_account() # 获取本月汇总 now = datetime.now() monthly_summary = self.storage.get_monthly_summary(now.year, now.month) # 获取最近交易 recent_transactions = self.storage.transactions.get_transactions(limit=5) recent_data = [] for t in recent_transactions: recent_data.append({ "id": t.id, "amount": float(t.amount), "category": t.category, "description": t.description, "date": t.date.isoformat(), "type": t.transaction_type.value }) # 获取分类信息用于统计 categories = self.storage.categories.load_categories() category_names = {cat.id: cat.name for cat in categories} return { "success": True, "content": json.dumps({ "account_summary": { "balance": float(account.balance), "total_transactions": account.total_transactions, "last_updated": account.last_updated.isoformat() if account.last_updated else None, "created_at": account.created_at.isoformat() if account.created_at else None }, "monthly_summary": { "year": now.year, "month": now.month, "month_name": now.strftime("%B"), "total_income": monthly_summary["total_income"], "total_expense": monthly_summary["total_expense"], "net_flow": monthly_summary["net_flow"], "transaction_count": monthly_summary["transaction_count"], "category_breakdown": monthly_summary["category_breakdown"] }, "recent_transactions": recent_data, "quick_stats": { "avg_transaction": float(account.balance / account.total_transactions) if account.total_transactions > 0 else 0, "transactions_this_month": monthly_summary["transaction_count"], "days_since_last_transaction": (datetime.now().date() - recent_transactions[0].date).days if recent_transactions else None }, "metadata": { "generated_at": datetime.now().isoformat(), "currency": "CNY", "timezone": "Asia/Shanghai" } }, indent=2, ensure_ascii=False), "mimeType": "application/json" } def get_resource_definitions() -> List[Dict[str, Any]]: """返回所有资源的 MCP 定义""" return [ { "uri": "transactions://all", "name": "所有交易记录", "description": "包含所有交易记录的完整数据集，包括统计元数据", "mimeType": "application/json" }, { "uri": "categories://list", "name": "分类列表", "description": "所有可用的收入和支出分类定义，包括图标和颜色信息", "mimeType": "application/json" }, { "uri": "summary://current", "name": "当前汇总", "description": "账户余额、本月统计、最近交易等汇总信息", "mimeType": "application/json" } ]