get_transactions

Retrieve transactions from Monarch Money, filtering by date range, account ID, and pagination. Specify limits and offsets to manage data retrieval efficiently.

Instructions

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

Input Schema

TableJSON Schema

Name	Required	Description	Default
`account_id`	No
`end_date`	No
`limit`	No
`offset`	No
`start_date`	No

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")

monarch-mcp-server