get_budgets
Retrieve budget data from Monarch Money to manage and track financial goals effectively using the Monarch MCP Server.
Instructions
Get budget information from Monarch Money.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- src/monarch_mcp_server/server.py:247-275 (handler)The primary handler function for the 'get_budgets' MCP tool. It is registered via the @mcp.tool() decorator, fetches an authenticated MonarchMoney client, calls the library's get_budgets() method asynchronously (using run_async wrapper), formats the response into a JSON list of budget details, and handles errors.@mcp.tool() def get_budgets() -> str: """Get budget information from Monarch Money.""" try: async def _get_budgets(): client = await get_monarch_client() return await client.get_budgets() budgets = run_async(_get_budgets()) # Format budgets for display budget_list = [] for budget in budgets.get("budgets", []): budget_info = { "id": budget.get("id"), "name": budget.get("name"), "amount": budget.get("amount"), "spent": budget.get("spent"), "remaining": budget.get("remaining"), "category": budget.get("category", {}).get("name"), "period": budget.get("period"), } budget_list.append(budget_info) return json.dumps(budget_list, indent=2, default=str) except Exception as e: logger.error(f"Failed to get budgets: {e}") return f"Error getting budgets: {str(e)}"
- Helper function used by get_budgets (and other tools) to obtain an authenticated MonarchMoney client instance, either from secure session storage or by logging in with environment variables.async def get_monarch_client() -> MonarchMoney: """Get or create MonarchMoney client instance using secure session storage.""" # Try to get authenticated client from secure session client = secure_session.get_authenticated_client() if client is not None: logger.info("✅ Using authenticated client from secure keyring storage") return client # If no secure session, try environment credentials email = os.getenv("MONARCH_EMAIL") password = os.getenv("MONARCH_PASSWORD") if email and password: try: client = MonarchMoney() await client.login(email, password) logger.info( "Successfully logged into Monarch Money with environment credentials" ) # Save the session securely secure_session.save_authenticated_session(client) return client except Exception as e: logger.error(f"Failed to login to Monarch Money: {e}") raise raise RuntimeError("🔐 Authentication needed! Run: python login_setup.py")
- Helper utility to execute async coroutines synchronously from the sync tool handler context, specifically used to run the internal _get_budgets() async function.def run_async(coro): """Run async function in a new thread with its own event loop.""" def _run(): loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: return loop.run_until_complete(coro) finally: loop.close() with ThreadPoolExecutor() as executor: future = executor.submit(_run) return future.result()