YNAB MCP Server

Instructions

Args:
    budget_id: The ID of the budget (use 'last-used' for default budget)

Returns:
    JSON string with list of scheduled transactions

Implementation Reference

• src/ynab_mcp/server.py:343-356 (handler)

  MCP tool handler for get_scheduled_transactions. Registers the tool and delegates to YNABClient instance.

  @mcp.tool()
  async def get_scheduled_transactions(budget_id: str) -> str:
      """Get all scheduled transactions.
  
      Args:
          budget_id: The ID of the budget (use 'last-used' for default budget)
  
      Returns:
          JSON string with list of scheduled transactions
      """
      client = get_ynab_client()
      result = await client.get_scheduled_transactions(budget_id)
      return json.dumps(result, indent=2)

• src/ynab_mcp/ynab_client.py:1084-1122 (helper)

  Core implementation in YNABClient that fetches scheduled transactions from YNAB API, formats the response, and handles errors.

  async def get_scheduled_transactions(self, budget_id: str) -> list[dict[str, Any]]:
      """Get all scheduled transactions.
  
      Args:
          budget_id: The budget ID or 'last-used'
  
      Returns:
          List of scheduled transaction dictionaries
      """
      try:
          url = f"{self.api_base_url}/budgets/{budget_id}/scheduled_transactions"
  
          result = await self._make_request_with_retry("get", url)
  
          scheduled_txns = []
          for txn in result["data"]["scheduled_transactions"]:
              scheduled_txns.append(
                  {
                      "id": txn["id"],
                      "date_first": txn.get("date_first"),
                      "date_next": txn.get("date_next"),
                      "frequency": txn.get("frequency"),
                      "amount": txn["amount"] / 1000 if txn.get("amount") else 0,
                      "memo": txn.get("memo"),
                      "flag_color": txn.get("flag_color"),
                      "account_id": txn.get("account_id"),
                      "account_name": txn.get("account_name"),
                      "payee_id": txn.get("payee_id"),
                      "payee_name": txn.get("payee_name"),
                      "category_id": txn.get("category_id"),
                      "category_name": txn.get("category_name"),
                      "deleted": txn.get("deleted"),
                  }
              )
  
          return scheduled_txns
      except Exception as e:
          raise Exception(f"Failed to get scheduled transactions: {e}") from e