monarch-mcp-server

Instructions

Implementation Reference

• src/monarch_mcp_server/server.py:186-244 (handler)

  The primary handler function for the 'get_transactions' tool. Decorated with @mcp.tool() for registration in the MCP server. Handles input parameters, asynchronously fetches transactions from the MonarchMoney client with optional filters, formats them into a JSON list with key details (id, date, amount, etc.), and returns the result or error message.
  @mcp.tool() def get_transactions( limit: int = 100, offset: int = 0, start_date: Optional[str] = None, end_date: Optional[str] = None, account_id: Optional[str] = None, ) -> str: """ Get transactions from Monarch Money. Args: limit: Number of transactions to retrieve (default: 100) offset: Number of transactions to skip (default: 0) start_date: Start date in YYYY-MM-DD format end_date: End date in YYYY-MM-DD format account_id: Specific account ID to filter by """ try: async def _get_transactions(): client = await get_monarch_client() # Build filters filters = {} if start_date: filters["start_date"] = start_date if end_date: filters["end_date"] = end_date if account_id: filters["account_id"] = account_id return await client.get_transactions(limit=limit, offset=offset, **filters) transactions = run_async(_get_transactions()) # Format transactions for display transaction_list = [] for txn in transactions.get("allTransactions", {}).get("results", []): transaction_info = { "id": txn.get("id"), "date": txn.get("date"), "amount": txn.get("amount"), "description": txn.get("description"), "category": txn.get("category", {}).get("name") if txn.get("category") else None, "account": txn.get("account", {}).get("displayName"), "merchant": txn.get("merchant", {}).get("name") if txn.get("merchant") else None, "is_pending": txn.get("isPending", False), } transaction_list.append(transaction_info) return json.dumps(transaction_list, indent=2, default=str) except Exception as e: logger.error(f"Failed to get transactions: {e}") return f"Error getting transactions: {str(e)}"
• src/monarch_mcp_server/server.py:31-45 (helper)

  Helper function used by get_transactions (and other tools) to execute async coroutines synchronously within the MCP server context, creating a new event loop in a thread pool.
  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()
• src/monarch_mcp_server/server.py:57-86 (helper)

  Helper function called within get_transactions to obtain an authenticated MonarchMoney client instance, either from secure session storage or environment credentials.
  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")